我正在测试 InfluxDB 来存储传感器时间序列。我正在使用 influxdb-java 客户端库(版本 2.15)，并且我在本地运行 InfluxDB 1.7.6 进行测试。
我所有的点都存储在.csv文件(每个传感器一个)，这些文件本身存储在.zip文件中(每个数据集一个)。我的代码贯穿每个 csv 文件的每一行。点以批处理模式写入。
/**
* Get the connection to the database
*/
InfluxDB influxDB = InfluxDBFactory.connect("http://192.168.51.51:8086");
influxDB.query(new Query("CREATE DATABASE theia_in_situ"));
influxDB.setDatabase("theia_in_situ");
influxDB.enableBatch();
influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
/**
* Create batch point to write each measure of the time serie more efficiently
*/
BatchPoints batchPoints = BatchPoints
.database("theia_in_situ")
.build();
对于每个 CSV 数据文件，将执行以下方法：
public static void createAndImportTimeSeriesDocuments(InputStream txtFileIn, String observationId, String producerId,
InfluxDB influxDB, BatchPoints batchPoints) throws IOException, ParseException {
DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
/**
* Store the variable name
*/
String observedProperty = null;
try (BufferedReader br = new BufferedReader(new InputStreamReader(txtFileIn));) {
String line = null;
/**
* Read the headers
*/
while ((line = br.readLine()).substring(0, 1).equals("#")) {
if (line.substring(0, 15).equals("#Variable_name;")) {
observedProperty = line.split(";")[1];
}
}
/**
* Read the data
*/
while ((line = br.readLine()) != null) {
String[] lineSplitted = line.split(";", -1);
Point point = Point.measurement(observedProperty)
.tag("producerId", producerId)
.tag("observationId", observationId)
.time(df.parse(lineSplitted[1]).getTime(), TimeUnit.MILLISECONDS)
.addField("value", lineSplitted[5])
.addField("flag", lineSplitted[6])
.build();
batchPoints.point(point);
}
influxDB.write(batchPoints);
}
}
我可以写一个或几个测量值，但很快我就得到了以下异常：
线程"main"中的异常 org.influxdb.InfluxDBIOException： java.net.SocketException： 对等方重置连接： 套接字写入错误 at org.influxdb.impl.InfluxDBImpl.execute(InfluxDBImpl.java：812) at org.influxdb.impl.InfluxDBImpl.write(InfluxDBImpl.java：463)
我已经禁用了最大并发写入限制，最大排队写入限制，排队写入超时(/etc/influxdb/influxdb.conf中将每个值设置为0)，如此处所述。 尽管这个问题在 Github 页面中被提及为常见问题解答，但我找不到任何问题来重现我的问题。
任何帮助将不胜感激。
尝试以批处理模式写入BatchPoint时似乎会发生此异常。
influxdb-java 客户端正在将您的写入存储到内部 缓冲区并以固定刷新的方式将它们异步刷新到 InfluxDB 间隔，以在客户端和服务器端实现良好的性能。
这是更新的代码段。
/**
* Read the data
*/
while ((line = br.readLine()) != null) {
String[] lineSplitted = line.split(";", -1);
Point point = Point.measurement(observedProperty)
.tag("producerId", producerId)
.tag("observationId", observationId)
.time(df.parse(lineSplitted[1]).getTime(), TimeUnit.MILLISECONDS)
.addField("value", lineSplitted[5])
.addField("flag", lineSplitted[6])
.build();
influxDB.write(point);
// batchPoints.point(point);
}
//influxDB.write(batchPoints);