<dependencies>
    <dependency>
      <groupId>org.apache.iotdb</groupId>
      <artifactId>iotdb-session</artifactId>
      <!-- 版本號(hào)與數(shù)據(jù)庫版本號(hào)相同 -->
      <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) {
        // 配置數(shù)據(jù)庫節(jié)點(diǎn)URL，支持多節(jié)點(diǎn)容災(zāi)
        List<String> nodeUrls = new ArrayList<>();
        nodeUrls.add("127.0.0.1:6667");
        nodeUrls.add("127.0.0.1:6668");
        // 構(gòu)建SessionPool
        sessionPool =
                new SessionPool.Builder()
                        .nodeUrls(nodeUrls) // 節(jié)點(diǎn)列表
                        .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. 執(zhí)行批量插入
        insertRecordsExample();
          // 3. 關(guān)閉連接池
        closeSessionPool();
  }

  // 構(gòu)建連接池
  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();
    }

    // 批量插入多行數(shù)據(jù)
    public static void insertRecordsExample() throws IoTDBConnectionException, StatementExecutionException {
        String deviceId = "root.sg1.d1";
        // 定義測(cè)點(diǎn)列表
        List<String> measurements = new ArrayList<>();
        measurements.add("s1");
        measurements.add("s2");
        measurements.add("s3");

        // 初始化批量寫入的參數(shù)集合
        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條數(shù)據(jù)，每100條批量寫入一次
        for (long time = 0; time < 500; time++) {
            List<Object> values = new ArrayList<>();
            List<TSDataType> types = new ArrayList<>();
            // 模擬測(cè)點(diǎn)值
            values.add(1L);
            values.add(2L);
            values.add(3L);
            // 定義測(cè)點(diǎn)數(shù)據(jù)類型
            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條執(zhí)行一次寫入，避免集合過大
            if (time != 0 && time % 100 == 0) {
                try {
                    sessionPool.insertRecords(deviceIds, timestamps, measurementsList, typesList, valuesList);
                } catch (IoTDBConnectionException | StatementExecutionException e) {
                    // 異常處理，如重試、日志記錄
                }
                // 清空集合，準(zhǔn)備下一批數(shù)據(jù)
                deviceIds.clear();
                measurementsList.clear();
                valuesList.clear();
                typesList.clear();
                timestamps.clear();
            }
        }
        // 寫入最后一批不足100條的數(shù)據(jù)
        try {
            sessionPool.insertRecords(deviceIds, timestamps, measurementsList, typesList, valuesList);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            // 異常處理
        }
    }
    // 關(guān)閉連接池
    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. 執(zhí)行Tablet寫入
        insertTabletExample();
        // 3. 關(guān)閉連接池
        closeSessionPool();
    }

    // 構(gòu)建連接池
    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();
    }

    // 單設(shè)備Tablet高效寫入
    private static void insertTabletExample() throws IoTDBConnectionException, StatementExecutionException {
        // 定義設(shè)備的測(cè)點(diǎn)模式（測(cè)點(diǎn)名稱+數(shù)據(jù)類型）
        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));

        // 創(chuàng)建Tablet對(duì)象：設(shè)備ID + 測(cè)點(diǎn)模式 + 最大行數(shù)量
        Tablet tablet = new Tablet("root.sg.d1",schemaList,100);

        // 模擬生成數(shù)據(jù)并寫入Tablet
        long timestamp = System.currentTimeMillis();
        Random random = new Random();
        for (long row = 0; row < 100; row++) {
            int rowIndex = tablet.getRowSize();
            // 添加時(shí)間戳
            tablet.addTimestamp(rowIndex, timestamp);
            // 為每個(gè)測(cè)點(diǎn)添加值
            for (int s = 0; s < 3; s++) {
                long value = random.nextLong();
                tablet.addValue(schemaList.get(s).getMeasurementName(), rowIndex, value);
            }
            // 當(dāng)Tablet達(dá)到最大行數(shù)量時(shí)，執(zhí)行寫入并重置
            if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
                sessionPool.insertTablet(tablet);
                tablet.reset();
            }
            timestamp++;
        }
        // 寫入最后一批不足最大行數(shù)量的數(shù)據(jù)
        if (tablet.getRowSize() != 0) {
            sessionPool.insertTablet(tablet);
            tablet.reset();
        }
    }

    // 關(guān)閉連接池
    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. 執(zhí)行非查詢SQL（DDL/DML）
        executeNonQueryExample();
        // 3. 執(zhí)行查詢SQL，遍歷結(jié)果集
        executeQueryExample();
        // 4. 關(guān)閉連接池
        closeSessionPool();
    }

    // 非查詢SQL操作：創(chuàng)建時(shí)間序列、設(shè)置TTL、刪除時(shí)間序列等
    private static void executeNonQueryExample() throws IoTDBConnectionException, StatementExecutionException {
        // 創(chuàng)建非對(duì)齊時(shí)間序列
        sessionPool.executeNonQueryStatement("create timeseries root.test.d1.s1 with dataType = int32");
        // 為指定路徑設(shè)置TTL（過期時(shí)間）
        sessionPool.executeNonQueryStatement("set TTL to root.test.** 10000");
        // 刪除時(shí)間序列
        sessionPool.executeNonQueryStatement("delete timeseries root.test.d1.s1");
    }

    // 查詢SQL操作：普通查詢 + 聚合查詢，遍歷結(jié)果集
    private static void executeQueryExample() throws IoTDBConnectionException, StatementExecutionException {
        // 1. 普通查詢：查詢前10條數(shù)據(jù)
        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());
            // 遍歷結(jié)果集
            while (dataIterator.next()) {
                StringBuilder builder = new StringBuilder();
                for (String columnName : wrapper.getColumnNames()) {
                    builder.append(dataIterator.getString(columnName) + " ");
                }
                System.out.println(builder);
            }
        }

        // 2. 聚合查詢：按5ms窗口分組統(tǒng)計(jì)s1的數(shù)量
        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);
            }
        }
    }

    // 構(gòu)建連接池
    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();
    }

    // 關(guān)閉連接池
    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. 執(zhí)行查詢并遍歷多類型結(jié)果集
    executeQueryExample();
    // 3. 關(guān)閉連接池
    closeSessionPool();
  }

  private static void executeQueryExample()
      throws IoTDBConnectionException, StatementExecutionException {
    // 先插入一條多數(shù)據(jù)類型的測(cè)試數(shù)據(jù)
    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); // 標(biāo)記s11為null
    tablet.rowSize = 1;
    sessionPool.insertAlignedTablet(tablet);

    // 查詢并遍歷多類型結(jié)果集
    try (SessionDataSetWrapper dataSet =
        sessionPool.executeQueryStatement("select * from root.sg.d1")) {
      SessionDataSet.DataIterator iterator = dataSet.iterator();
      int count = 0;
      while (iterator.next()) {
        count++;
        // 各種數(shù)據(jù)類型的取值 + 空值判斷
        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")); // 長(zhǎng)整型

        Assert.assertFalse(iterator.isNull("root.sg.d1.s3"));
        Assert.assertEquals(0, iterator.getFloat("root.sg.d1.s3"), 0.01); // 浮點(diǎn)型

        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")); // 時(shí)間戳

        Assert.assertFalse(iterator.isNull("root.sg.d1.s8"));
        Assert.assertEquals(new Binary(new byte[] {1}), iterator.getBlob("root.sg.d1.s8")); // 二進(jìn)制

        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();
  }
}