Java开发者必看：IoTDB原生接口SessionPool 的高效使用与避坑指南

<dependencies>
    <dependency>
      <groupid>org.apache.iotdb</groupid>
      <artifactid>iotdb-session</artifactid>
      <!-- 版本号与数据库版本号相同 -->
      <version>${project.version}</version>
    </dependency>
</dependencies>

import java.util.arraylist;
import java.util.list;
import org.apache.iotdb.session.pool.sessionpool;

public class iotdbsessionpoolexample {
    private static sessionpool sessionpool;

    public static void main(string[] args) {
        // 配置数据库节点url，支持多节点容灾
        list<string> nodeurls = new arraylist<>();
        nodeurls.add("127.0.0.1:6667");
        nodeurls.add("127.0.0.1:6668");
        // 构建sessionpool
        sessionpool =
                new sessionpool.builder()
                        .nodeurls(nodeurls) // 节点列表
                        .user("root")       // 用户名
                        .password("root")   // 密码
                        .maxsize(3)         // 连接池最大大小
                        .build();
    }
}

import java.util.arraylist;
import java.util.list;
import org.apache.iotdb.rpc.iotdbconnectionexception;
import org.apache.iotdb.rpc.statementexecutionexception;
import org.apache.iotdb.session.pool.sessionpool;
import org.apache.tsfile.enums.tsdatatype;

public class sessionpoolexample {
 private static sessionpool sessionpool;

     public static void main(string[] args) throws iotdbconnectionexception, statementexecutionexception {
          // 1. 初始化连接池
        constructsessionpool();
          // 2. 执行批量插入
        insertrecordsexample();
          // 3. 关闭连接池
        closesessionpool();
  }

  // 构建连接池
  private static void constructsessionpool() {
        list<string> nodeurls = new arraylist<>();
        nodeurls.add("127.0.0.1:6667");
        nodeurls.add("127.0.0.1:6668");
        sessionpool =
                new sessionpool.builder()
                        .nodeurls(nodeurls)
                        .user("root")
                        .password("root")
                        .maxsize(3)
                        .build();
    }

    // 批量插入多行数据
    public static void insertrecordsexample() throws iotdbconnectionexception, statementexecutionexception {
        string deviceid = "root.sg1.d1";
        // 定义测点列表
        list<string> measurements = new arraylist<>();
        measurements.add("s1");
        measurements.add("s2");
        measurements.add("s3");

        // 初始化批量写入的参数集合
        list<string> deviceids = new arraylist<>();
        list<list<string>> measurementslist = new arraylist<>();
        list<list<object>> valueslist = new arraylist<>();
        list<long> timestamps = new arraylist<>();
        list<list<tsdatatype>> typeslist = new arraylist<>();

        // 模拟生成500条数据，每100条批量写入一次
        for (long time = 0; time < 500; time++) {
            list<object> values = new arraylist<>();
            list<tsdatatype> types = new arraylist<>();
            // 模拟测点值
            values.add(1l);
            values.add(2l);
            values.add(3l);
            // 定义测点数据类型
            types.add(tsdatatype.int64);
            types.add(tsdatatype.int64);
            types.add(tsdatatype.int64);
            // 添加到批量集合
            deviceids.add(deviceid);
            measurementslist.add(measurements);
            valueslist.add(values);
            typeslist.add(types);
            timestamps.add(time);

            // 每100条执行一次写入，避免集合过大
            if (time != 0 && time % 100 == 0) {
                try {
                    sessionpool.insertrecords(deviceids, timestamps, measurementslist, typeslist, valueslist);
                } catch (iotdbconnectionexception | statementexecutionexception e) {
                    // 异常处理，如重试、日志记录
                }
                // 清空集合，准备下一批数据
                deviceids.clear();
                measurementslist.clear();
                valueslist.clear();
                typeslist.clear();
                timestamps.clear();
            }
        }
        // 写入最后一批不足100条的数据
        try {
            sessionpool.insertrecords(deviceids, timestamps, measurementslist, typeslist, valueslist);
        } catch (iotdbconnectionexception | statementexecutionexception e) {
            // 异常处理
        }
    }
    // 关闭连接池
    public static void closesessionpool(){
        sessionpool.close();
    }
}

import java.util.arraylist;
import java.util.list;
import java.util.random;
import org.apache.iotdb.rpc.iotdbconnectionexception;
import org.apache.iotdb.rpc.statementexecutionexception;
import org.apache.iotdb.session.pool.sessionpool;
import org.apache.tsfile.enums.tsdatatype;
import org.apache.tsfile.write.record.tablet;
import org.apache.tsfile.write.schema.imeasurementschema;
import org.apache.tsfile.write.schema.measurementschema;

public class sessionpoolexample {
    private static sessionpool sessionpool;

    public static void main(string[] args) throws iotdbconnectionexception, statementexecutionexception {
        // 1. 初始化连接池
        constructsessionpool();
        // 2. 执行tablet写入
        inserttabletexample();
        // 3. 关闭连接池
        closesessionpool();
    }

    // 构建连接池
    private static void constructsessionpool() {
        list<string> nodeurls = new arraylist<>();
        nodeurls.add("127.0.0.1:6667");
        sessionpool =
                new sessionpool.builder()
                        .nodeurls(nodeurls)
                        .user("root")
                        .password("root")
                        .maxsize(3)
                        .build();
    }

    // 单设备tablet高效写入
    private static void inserttabletexample() throws iotdbconnectionexception, statementexecutionexception {
        // 定义设备的测点模式（测点名称+数据类型）
        list<imeasurementschema> schemalist = new arraylist<>();
        schemalist.add(new measurementschema("s1", tsdatatype.int64));
        schemalist.add(new measurementschema("s2", tsdatatype.int64));
        schemalist.add(new measurementschema("s3", tsdatatype.int64));

        // 创建tablet对象：设备id + 测点模式 + 最大行数量
        tablet tablet = new tablet("root.sg.d1",schemalist,100);

        // 模拟生成数据并写入tablet
        long timestamp = system.currenttimemillis();
        random random = new random();
        for (long row = 0; row < 100; row++) {
            int rowindex = tablet.getrowsize();
            // 添加时间戳
            tablet.addtimestamp(rowindex, timestamp);
            // 为每个测点添加值
            for (int s = 0; s < 3; s++) {
                long value = random.nextlong();
                tablet.addvalue(schemalist.get(s).getmeasurementname(), rowindex, value);
            }
            // 当tablet达到最大行数量时，执行写入并重置
            if (tablet.getrowsize() == tablet.getmaxrownumber()) {
                sessionpool.inserttablet(tablet);
                tablet.reset();
            }
            timestamp++;
        }
        // 写入最后一批不足最大行数量的数据
        if (tablet.getrowsize() != 0) {
            sessionpool.inserttablet(tablet);
            tablet.reset();
        }
    }

    // 关闭连接池
    public static void closesessionpool(){
        sessionpool.close();
    }
}

import java.util.arraylist;
import java.util.list;
import org.apache.iotdb.isession.pool.sessiondatasetwrapper;
import org.apache.iotdb.rpc.iotdbconnectionexception;
import org.apache.iotdb.rpc.statementexecutionexception;
import org.apache.iotdb.session.pool.sessionpool;
import org.apache.iotdb.isession.sessiondataset.dataiterator;

public class sessionpoolexample {
  private static sessionpool sessionpool;

  public static void main(string[] args) throws iotdbconnectionexception, statementexecutionexception {
        // 1. 初始化连接池
        constructsessionpool();
        // 2. 执行非查询sql（ddl/dml）
        executenonqueryexample();
        // 3. 执行查询sql，遍历结果集
        executequeryexample();
        // 4. 关闭连接池
        closesessionpool();
    }

    // 非查询sql操作：创建时间序列、设置ttl、删除时间序列等
    private static void executenonqueryexample() throws iotdbconnectionexception, statementexecutionexception {
        // 创建非对齐时间序列
        sessionpool.executenonquerystatement("create timeseries root.test.d1.s1 with datatype = int32");
        // 为指定路径设置ttl（过期时间）
        sessionpool.executenonquerystatement("set ttl to root.test.** 10000");
        // 删除时间序列
        sessionpool.executenonquerystatement("delete timeseries root.test.d1.s1");
    }

    // 查询sql操作：普通查询 + 聚合查询，遍历结果集
    private static void executequeryexample() throws iotdbconnectionexception, statementexecutionexception {
        // 1. 普通查询：查询前10条数据
        try(sessiondatasetwrapper wrapper = sessionpool.executequerystatement("select s1 from root.sg1.d1 limit 10")) {
            dataiterator dataiterator = wrapper.iterator();
            // 打印列名和列类型
            system.out.println(wrapper.getcolumnnames());
            system.out.println(wrapper.getcolumntypes());
            // 遍历结果集
            while (dataiterator.next()) {
                stringbuilder builder = new stringbuilder();
                for (string columnname : wrapper.getcolumnnames()) {
                    builder.append(dataiterator.getstring(columnname) + " ");
                }
                system.out.println(builder);
            }
        }

        // 2. 聚合查询：按5ms窗口分组统计s1的数量
        try(sessiondatasetwrapper wrapper = sessionpool.executequerystatement("select count(s1) from root.sg1.d1 group by ([0, 40), 5ms) ")) {
            dataiterator dataiterator = wrapper.iterator();
            system.out.println(wrapper.getcolumnnames());
            system.out.println(wrapper.getcolumntypes());
            while (dataiterator.next()) {
                stringbuilder builder = new stringbuilder();
                for (string columnname : wrapper.getcolumnnames()) {
                    builder.append(dataiterator.getstring(columnname) + " ");
                }
                system.out.println(builder);
            }
        }
    }

    // 构建连接池
    private static void constructsessionpool() {
        list<string> nodeurls = new arraylist<>();
        nodeurls.add("127.0.0.1:6667");
        nodeurls.add("127.0.0.1:6668");
        sessionpool =
                new sessionpool.builder()
                        .nodeurls(nodeurls)
                        .user("root")
                        .password("root")
                        .maxsize(3)
                        .build();
    }

    // 关闭连接池
    public static void closesessionpool(){
        sessionpool.close();
    }
}

import org.apache.iotdb.isession.sessiondataset;
import org.apache.iotdb.isession.pool.sessiondatasetwrapper;
import org.apache.iotdb.rpc.iotdbconnectionexception;
import org.apache.iotdb.rpc.statementexecutionexception;
import org.apache.iotdb.session.pool.sessionpool;
import org.apache.tsfile.enums.tsdatatype;
import org.apache.tsfile.utils.binary;
import org.apache.tsfile.utils.dateutils;
import org.apache.tsfile.write.record.tablet;
import org.apache.tsfile.write.schema.measurementschema;
import org.junit.assert;
import java.sql.timestamp;
import java.util.arraylist;
import java.util.arrays;
import java.util.list;

public class sessionexample {
  private static sessionpool sessionpool;

  public static void main(string[] args)
      throws iotdbconnectionexception, statementexecutionexception {
    // 1. 初始化连接池
    constructsessionpool();
    // 2. 执行查询并遍历多类型结果集
    executequeryexample();
    // 3. 关闭连接池
    closesessionpool();
  }

  private static void executequeryexample()
      throws iotdbconnectionexception, statementexecutionexception {
    // 先插入一条多数据类型的测试数据
    tablet tablet =
        new tablet(
            "root.sg.d1",
            arrays.aslist(
                new measurementschema("s1", tsdatatype.int32),
                new measurementschema("s2", tsdatatype.int64),
                new measurementschema("s3", tsdatatype.float),
                new measurementschema("s4", tsdatatype.double),
                new measurementschema("s5", tsdatatype.text),
                new measurementschema("s6", tsdatatype.boolean),
                new measurementschema("s7", tsdatatype.timestamp),
                new measurementschema("s8", tsdatatype.blob),
                new measurementschema("s9", tsdatatype.string),
                new measurementschema("s10", tsdatatype.date),
                new measurementschema("s11", tsdatatype.timestamp)),
            10);
    tablet.addtimestamp(0, 0l);
    tablet.addvalue("s1", 0, 1);
    tablet.addvalue("s2", 0, 1l);
    tablet.addvalue("s3", 0, 0f);
    tablet.addvalue("s4", 0, 0d);
    tablet.addvalue("s5", 0, "text_value");
    tablet.addvalue("s6", 0, true);
    tablet.addvalue("s7", 0, 1l);
    tablet.addvalue("s8", 0, new binary(new byte[] {1}));
    tablet.addvalue("s9", 0, "string_value");
    tablet.addvalue("s10", 0, dateutils.parseinttolocaldate(20250403));
    tablet.initbitmaps();
    tablet.bitmaps[10].mark(0); // 标记s11为null
    tablet.rowsize = 1;
    sessionpool.insertalignedtablet(tablet);

    // 查询并遍历多类型结果集
    try (sessiondatasetwrapper dataset =
        sessionpool.executequerystatement("select * from root.sg.d1")) {
      sessiondataset.dataiterator iterator = dataset.iterator();
      int count = 0;
      while (iterator.next()) {
        count++;
        // 各种数据类型的取值 + 空值判断
        assert.assertfalse(iterator.isnull("root.sg.d1.s1"));
        assert.assertequals(1, iterator.getint("root.sg.d1.s1")); // 整型

        assert.assertfalse(iterator.isnull("root.sg.d1.s2"));
        assert.assertequals(1l, iterator.getlong("root.sg.d1.s2")); // 长整型

        assert.assertfalse(iterator.isnull("root.sg.d1.s3"));
        assert.assertequals(0, iterator.getfloat("root.sg.d1.s3"), 0.01); // 浮点型

        assert.assertfalse(iterator.isnull("root.sg.d1.s4"));
        assert.assertequals(0, iterator.getdouble("root.sg.d1.s4"), 0.01); // 双精度

        assert.assertfalse(iterator.isnull("root.sg.d1.s5"));
        assert.assertequals("text_value", iterator.getstring("root.sg.d1.s5")); // 文本

        assert.assertfalse(iterator.isnull("root.sg.d1.s6"));
        assert.asserttrue(iterator.getboolean("root.sg.d1.s6")); // 布尔值

        assert.assertfalse(iterator.isnull("root.sg.d1.s7"));
        assert.assertequals(new timestamp(1), iterator.gettimestamp("root.sg.d1.s7")); // 时间戳

        assert.assertfalse(iterator.isnull("root.sg.d1.s8"));
        assert.assertequals(new binary(new byte[] {1}), iterator.getblob("root.sg.d1.s8")); // 二进制

        assert.assertfalse(iterator.isnull("root.sg.d1.s10"));
        assert.assertequals(dateutils.parseinttolocaldate(20250403), iterator.getdate("root.sg.d1.s10")); // 日期

        assert.asserttrue(iterator.isnull("root.sg.d1.s11")); // 空值判断
        assert.assertnull(iterator.gettimestamp("root.sg.d1.s11"));
      }
      assert.assertequals(tablet.rowsize, count);
    }
  }

  private static void constructsessionpool() {
    list<string> nodeurls = new arraylist<>();
    nodeurls.add("127.0.0.1:6667");
    sessionpool =
        new sessionpool.builder()
            .nodeurls(nodeurls)
            .user("root")
            .password("root")
            .maxsize(3)
            .build();
  }

  public static void closesessionpool() {
    sessionpool.close();
  }
}