Influxdb-java：org.influxdb.InfluxDBIOException：java.net.SocketException：通过对等方重置连接：套接字写入错误

Question

我正在测试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：由org.influxdb.impl上的org.influxdb.impl.InfluxDBImpl.execute（InfluxDBImpl.java:812）中的peer：socket write error重置连接。 InfluxDBImpl.write（InfluxDBImpl.java:463）

我已经禁用了max-concurrent-write-limit，max-enqueued-write-limit，enqueued-write-timeout（在/etc/influxdb/influxdb.conf中将每个值设置为0），如提到的here。即使在Github页面中提到这个问题作为FAQ，我也找不到任何问题来重现我的问题。

任何帮助，将不胜感激。

Answer 1

尝试在批处理模式下编写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);