SpringBoot集成时序数据库IoTDB(IotDB-Session方式)

阐明：项目需利用时序数据库，后经调研选用了IoTDB数据库，因此需要在SpringBoot框架内集成该数据库，来进行数据查询、插入操作。根据官方文档Java可采用多种方式访问数据库，分别是Java原生接口、JDBC(不保举)和REST API，下面选用Java原生接口和MyBatis的方式操作数据库。
现实应用中需要高性能吞吐则利用Java原生接口的方式，但一些小量级的增删查可以利用MyBatis的方式。
前提：已摆设安装IoTDB数据库，具体安装摆设流程可查察官网文档。
1、IotDB-Session

1.1、pom.xml

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.example</groupId>
8.     <artifactId>springboot-iotdb</artifactId>
9.     <version>1.0-SNAPSHOT</version>
11.     <properties>
12.         <maven.compiler.source>8</maven.compiler.source>
13.         <maven.compiler.target>8</maven.compiler.target>
14.     </properties>
15.     <parent>
16.         <artifactId>spring-boot-starter-parent</artifactId>
17.         <groupId>org.springframework.boot</groupId>
18.         <version>2.1.5.RELEASE</version>
19.     </parent>
21.     <dependencies>
22.         <dependency>
23.             <groupId>org.projectlombok</groupId>
24.             <artifactId>lombok</artifactId>
25.             <version>RELEASE</version>
26.             <scope>compile</scope>
27.         </dependency>
28.         <dependency>
29.             <groupId>cn.hutool</groupId>
30.             <artifactId>hutool-all</artifactId>
31.             <version>5.8.28</version>
32.             <scope>compile</scope>
33.         </dependency>
34.         <dependency>
35.             <groupId>org.springframework.boot</groupId>
36.             <artifactId>spring-boot-autoconfigure</artifactId>
37.         </dependency>
38.         <dependency>
39.             <groupId>org.springframework.boot</groupId>
40.             <artifactId>spring-boot-starter-web</artifactId>
41.         </dependency>
42.         <!-- LOG4J -->
43.         <dependency>
44.             <groupId>log4j</groupId>
45.             <artifactId>log4j</artifactId>
46.             <version>1.2.16</version>
47.             <scope>compile</scope>
48.         </dependency>
49.         <dependency>
50.             <groupId>org.springframework.boot</groupId>
51.             <artifactId>spring-boot-starter-test</artifactId>
52.             <scope>test</scope>
53.         </dependency>
55.         <dependency>
56.             <groupId>org.apache.iotdb</groupId>
57.             <artifactId>iotdb-session</artifactId>
58.             <version>1.3.1</version>
59.         </dependency>
61.         <dependency>
62.             <groupId>com.alibaba.fastjson2</groupId>
63.             <artifactId>fastjson2</artifactId>
64.             <version>2.0.51</version>
65.         </dependency>
66.     </dependencies>
68. </project>

复制代码

1.2、application.yml

1. spring:
2.   iotdb:
3.     username: root
4.     password: root
5.     ip: 100.100.100.100
6.     port: 6667
7.     maxSize: 100
8. server:
9.   port: 8080

复制代码

1.3、IotDbSessionConfig

1. package com.example.config;
3. import org.apache.iotdb.session.pool.SessionPool;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.beans.factory.annotation.Value;
7. import org.springframework.context.annotation.Configuration;
8. import org.springframework.stereotype.Component;
10. @Component
11. @Configuration
12. public class IotDbSessionConfig {
14.     @Value("${spring.iotdb.username}")
15.     private String username;
17.     @Value("${spring.iotdb.password}")
18.     private String password;
20.     @Value("${spring.iotdb.ip}")
21.     private String ip;
23.     @Value("${spring.iotdb.port}")
24.     private int port;
26.     @Value("${spring.iotdb.maxSize}")
27.     private int maxSize;
29.     private static SessionPool sessionPool;
31.     public SessionPool getSessionPool() {
32.         if (sessionPool == null) {
33.             sessionPool = new SessionPool(ip, port, username, password, maxSize);
34.         }
36.         return sessionPool;
37.     }
38. }

复制代码

1.4、IotDbService

1. package com.example.service;
3. import com.example.dto.IotDbRecordAble;
4. import com.example.dto.MeasurementSchemaValuesDTO;
5. import org.apache.iotdb.common.rpc.thrift.TAggregationType;
6. import org.apache.iotdb.isession.SessionDataSet;
7. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
8. import org.apache.iotdb.tsfile.write.record.Tablet;
9. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
11. import java.util.List;
12. import java.util.Map;
14. public interface IotDbService {
16.     void insertTablet(Tablet tablet);
18.     void insertTablets(Map<String, Tablet> tablets);
20.     void insertStringRecord(String deviceId, long time, List<String> measurements, List<String> values);
22.     void insertRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values);
24.     void insertStringRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList);
26.     void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList);
28.     void insertStringRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList);
30.     void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList);
32.     void deleteData(String path, long endTime);
34.     void deleteData(List<String> paths, long endTime);
36.     SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime, long timeOut);
38.     <T> List<T> executeRawDataQuery(List<String> paths, long startTime, long endTime, long timeOut, Class<? extends IotDbRecordAble> clazz);
40.     SessionDataSet executeLastDataQuery(List<String> paths, long lastTime);
42.     <T> List<T> executeLastDataQuery(List<String> paths, long lastTime, Class<? extends IotDbRecordAble> clazz);
44.     SessionDataSet executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes);
46.     <T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes, Class<? extends IotDbRecordAble> clazz);
48.     SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations);
50.     SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime);
52.     SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval);
54.     SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval, long slidingStep);
56.     SessionDataSet executeQueryStatement(String sql);
58.     void executeNonQueryStatement(String sql);
60.     List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> columnNames);
62.     /**
63.     *（不支持聚合查询）
64.     **/
65.     <T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList, Class<? extends IotDbRecordAble>  clazz);
67.     List<MeasurementSchema> buildMeasurementSchemas(Object object);
69.     MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object object);
70. }

复制代码

1.5、IotDbServiceImpl

1. package com.example.service.impl;
3. import com.alibaba.fastjson2.JSON;
4. import com.example.config.IotDbSessionConfig;
5. import com.example.dto.IotDbRecordAble;
6. import com.example.dto.MeasurementSchemaValuesDTO;
7. import com.example.service.IotDbService;
8. import lombok.SneakyThrows;
9. import lombok.extern.slf4j.Slf4j;
10. import org.apache.iotdb.common.rpc.thrift.TAggregationType;
11. import org.apache.iotdb.isession.SessionDataSet;
12. import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
13. import org.apache.iotdb.session.pool.SessionPool;
14. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
15. import org.apache.iotdb.tsfile.read.common.Field;
16. import org.apache.iotdb.tsfile.read.common.RowRecord;
17. import org.apache.iotdb.tsfile.write.record.Tablet;
18. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
19. import org.springframework.beans.factory.annotation.Autowired;
20. import org.springframework.stereotype.Service;
22. import java.lang.reflect.Type;
23. import java.util.*;
24. import java.util.stream.Collectors;
26. @Service
27. @Slf4j
28. public class IotDbServiceImpl implements IotDbService {
30.     @Autowired
31.     private IotDbSessionConfig iotDBSessionConfig;
33.     /**
34.      * 单设备批量插入数据
35.      *
36.      * @param tablet
37.      */
38.     @Override
39.     public void insertTablet(Tablet tablet) {
40.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
41.         try {
42.             log.info("iotdb数据入库：tablet:[{}]", tablet);
43.             sessionPool.insertTablet(tablet);
44.         } catch (Exception e) {
45.             log.error("IotDBSession insertTablet失败: tablet={},  error={}", tablet, e.getMessage());
46.         }
47.     }
49.     /**
50.      * 多设备批量插入数据
51.      *
52.      * @param tablets
53.      */
54.     @Override
55.     public void insertTablets(Map<String, Tablet> tablets) {
56.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
57.         try {
58.             log.info("iotdb数据入库：tablets:[{}]", tablets);
59.             sessionPool.insertTablets(tablets);
60.         } catch (Exception e) {
61.             log.error("IotDBSession insertTablets失败: tablets={},  error={}", tablets, e.getMessage());
62.         }
63.     }
65.     /**
66.      * 单条数据插入（string类型数据项）
67.      *
68.      * @param deviceId     设备名（表名）root.ln.wf01.wt01
69.      * @param time         时间戳
70.      * @param measurements 数据项列表
71.      * @param values       数据项对应值列表
72.      */
73.     @Override
74.     public void insertStringRecord(String deviceId, long time, List<String> measurements, List<String> values) {
75.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
76.         try {
77.             log.info("iotdb数据入库：device_id:[{}], measurements:[{}], values:[{}]", deviceId, measurements, values);
78.             sessionPool.insertRecord(deviceId, time, measurements, values);
79.         } catch (Exception e) {
80.             log.error("IotDBSession insertRecord失败: deviceId={}, time={}, measurements={}, values={}, error={}",
81.                     deviceId, time, measurements, values, e.getMessage());
82.         }
83.     }
85.     /**
86.      * 单条数据插入（不同类型数据项）
87.      *
88.      * @param deviceId     设备名（表名）root.ln.wf01.wt01
89.      * @param time         时间戳
90.      * @param measurements 数据项列表
91.      * @param types        数据项对应类型列表
92.      * @param values       数据项对应值列表
93.      */
94.     @Override
95.     public void insertRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values) {
96.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
97.         try {
98.             log.info("iotdb数据入库：device_id:[{}], measurements:[{}], types:[{}], values:[{}]", deviceId, measurements, types, values);
99.             sessionPool.insertRecord(deviceId, time, measurements, types, values);
100.         } catch (Exception e) {
101.             log.error("IotDBSession insertRecordHasTypes失败: deviceId={}, time={}, measurements={},types={}, values={}, error={}",
102.                     deviceId, time, measurements, types, values, e.getMessage());
103.         }
104.     }
107.     /**
108.      * 多个设备多条数据插入（string类型数据项）
109.      *
110.      * @param deviceIds        多个设备名（表名）root.ln.wf01.wt01
111.      * @param times            时间戳的列表
112.      * @param measurementsList 数据项列表的列表
113.      * @param valuesList       数据项对应值列表的列表
114.      */
115.     @Override
116.     public void insertStringRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
117.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
118.         try {
119.             log.info("iotdb数据入库：deviceIds:[{}], measurementsList:[{}], valuesList:[{}]", deviceIds, measurementsList, valuesList);
120.             sessionPool.insertRecords(deviceIds, times, measurementsList, valuesList);
121.         } catch (Exception e) {
122.             log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, valuesList={}, error={}",
123.                     deviceIds, times, measurementsList, valuesList, e.getMessage());
124.         }
125.     }
127.     /**
128.      * 多个设备多条数据插入（不同类型数据项）
129.      *
130.      * @param deviceIds        多个设备名（表名））root.ln.wf01.wt01
131.      * @param times            时间戳的列表
132.      * @param measurementsList 数据项列表的列表
133.      * @param typesList        数据项对应类型列表的列表
134.      * @param valuesList       数据项对应值列表的列表
135.      */
136.     @Override
137.     public void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
138.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
139.         try {
140.             log.info("iotdb数据入库：deviceIds:[{}], measurementsList:[{}], typesList：[{}], valuesList:[{}]", deviceIds, measurementsList, typesList, valuesList);
141.             sessionPool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList);
142.         } catch (Exception e) {
143.             log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}",
144.                     deviceIds, times, measurementsList, typesList, valuesList, e.getMessage());
145.         }
146.     }
148.     /**
149.      * 单个设备多条数据插入（string类型数据项）
150.      *
151.      * @param deviceId         单个设备名（表名））root.ln.wf01.wt01
152.      * @param times            时间戳的列表
153.      * @param measurementsList 数据项列表的列表
154.      * @param valuesList       数据项对应值列表的列表
155.      */
156.     @Override
157.     public void insertStringRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
158.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
159.         try {
160.             log.info("iotdb数据入库：deviceId:[{}], measurementsList:[{}], valuesList:[{}]", deviceId, measurementsList, valuesList);
161.             sessionPool.insertStringRecordsOfOneDevice(deviceId, times, measurementsList, valuesList);
162.         } catch (Exception e) {
163.             log.error("IotDBSession insertStringRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, valuesList={}, error={}",
164.                     deviceId, times, measurementsList, valuesList, e.getMessage());
165.         }
166.     }
168.     /**
169.      * 单个设备多条数据插入（不同类型数据项）
170.      *
171.      * @param deviceId         单个设备名（表名））root.ln.wf01.wt01
172.      * @param times            时间戳的列表
173.      * @param measurementsList 数据项列表的列表
174.      * @param typesList        数据项对应类型列表的列表
175.      * @param valuesList       数据项对应值列表的列表
176.      */
177.     @Override
178.     public void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
179.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
180.         try {
181.             log.info("iotdb数据入库：deviceId:[{}], measurementsList:[{}], typesList：[{}], valuesList:[{}]", deviceId, measurementsList, typesList, valuesList);
182.             sessionPool.insertRecordsOfOneDevice(deviceId, times, measurementsList, typesList, valuesList);
183.         } catch (Exception e) {
184.             log.error("IotDBSession insertRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}", deviceId, times, measurementsList, typesList, valuesList, e.getMessage());
185.         }
186.     }
188.     /**
189.      * 删除数据（删除一个时间序列在某个时间点前或这个时间点的数据）
190.      *
191.      * @param path    单个字段 root.ln.wf01.wt01.temperature
192.      * @param endTime 删除时间点
193.      */
194.     @Override
195.     public void deleteData(String path, long endTime) {
196.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
197.         try {
198.             log.info("iotdb数据删除：path:[{}], endTime:[{}]", path, endTime);
199.             sessionPool.deleteData(path, endTime);
200.         } catch (Exception e) {
201.             log.error("IotDBSession deleteData失败: deviceId={}, times={},error={}", path, endTime, e.getMessage());
202.         }
203.     }
205.     /**
206.      * 删除数据（删除多个时间序列在某个时间点前或这个时间点的数据）
207.      *
208.      * @param paths   多个字段（表名）） root.ln.wf01.wt01.temperature
209.      * @param endTime 删除时间点
210.      */
211.     @Override
212.     public void deleteData(List<String> paths, long endTime) {
213.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
214.         try {
215.             log.info("iotdb数据删除：paths:[{}], endTime:[{}]", paths, endTime);
216.             sessionPool.deleteData(paths, endTime);
217.         } catch (Exception e) {
218.             log.error("IotDBSession deleteData失败: paths={}, times={},error={}", paths, endTime, e.getMessage());
219.         }
220.     }
222.     /**
223.      * 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
224.      *
225.      * @param paths     多个字段（表名）） root.ln.wf01.wt01.temperature
226.      * @param startTime 开始时间
227.      * @param endTime   结束时间
228.      * @param outTime   超时时间
229.      * @return SessionDataSet  (Time,paths)
230.      */
231.     @Override
232.     public SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime) {
233.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
234.         SessionDataSetWrapper sessionDataSetWrapper = null;
235.         try {
236.             log.info("iotdb数据查询：paths:[{}], startTime：[{}], endTime:[{}],outTime:[{}]", paths, startTime, endTime, outTime);
237.             sessionDataSetWrapper = sessionPool.executeRawDataQuery(paths, startTime, endTime, outTime);
238.             return sessionDataSetWrapper.getSessionDataSet();
239.         } catch (Exception e) {
240.             log.error("IotDBSession executeRawDataQuery失败: paths={},  startTime：[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
241.         }
242.         return null;
243.     }
245.     /**
246.      * 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
247.      *
248.      * @param paths     多个字段（表名）） root.ln.wf01.wt01.temperature
249.      * @param startTime 开始时间
250.      * @param endTime   结束时间
251.      * @param outTime   超时时间
252.      * @param clazz     返回数据对应的对象（对象属性必须与字段名对应）
253.      * @param <T>
254.      * @return
255.      */
256.     @Override
257.     public <T> List<T> executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime, Class<? extends IotDbRecordAble> clazz) {
258.         SessionDataSet sessionDataSet = executeRawDataQuery(paths, startTime, endTime, outTime);
259.         List<String> columnNames = sessionDataSet.getColumnNames();
260.         List<T> resultEntities = null;
261.         try {
262.             resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
263.         } catch (Exception e) {
264.             log.error("IotDBSession executeRawDataQuery失败: paths={},  startTime：[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
265.         }
266.         return resultEntities;
267.     }
269.     /**
270.      * 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
271.      *
272.      * @param paths    多个字段（表名）） root.ln.wf01.wt01.temperature
273.      * @param lastTime 结束时间
274.      * @return SessionDataSet
275.      */
276.     @Override
277.     public SessionDataSet executeLastDataQuery(List<String> paths, long lastTime) {
278.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
279.         SessionDataSetWrapper sessionDataSetWrapper = null;
280.         try {
281.             log.info("iotdb数据查询：paths:[{}], lastTime:[{}]", paths, lastTime);
282.             sessionDataSetWrapper = sessionPool.executeLastDataQuery(paths, lastTime);
283.             return sessionDataSetWrapper.getSessionDataSet();
284.         } catch (Exception e) {
285.             log.error("IotDBSession executeLastDataQuery失败: paths={},  lastTime：[{}], error={}", paths, lastTime, e.getMessage());
286.         }
287.         return null;
288.     }
290.     /**
291.      * 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
292.      *
293.      * @param <T>
294.      * @param paths    多个字段（表名）） root.ln.wf01.wt01.temperature
295.      * @param lastTime 结束时间
296.      * @param clazz    返回数据对应的对象（对象属性必须与字段名对应）
297.      * @return
298.      */
299.     @Override
300.     public <T> List<T> executeLastDataQuery(List<String> paths, long lastTime, Class<? extends IotDbRecordAble> clazz) {
301.         SessionDataSet sessionDataSet = executeLastDataQuery(paths, lastTime);
302.         List<String> columnNames = sessionDataSet.getColumnNames();
303.         List<T> resultEntities = null;
304.         try {
305.             resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
306.         } catch (Exception e) {
307.             log.error("IotDBSession executeLastDataQuery失败: paths={},  lastTime：[{}], error={}", paths, lastTime, e.getMessage());
308.         }
309.         return resultEntities;
310.     }
312.     /**
313.      * 最新点查询（快速查询单设备下指定序列最新点）
314.      *
315.      * @param db               root.ln.wf01
316.      * @param device           root.ln.wf01.wt01
317.      * @param sensors          temperature，status(字段名)
318.      * @param isLegalPathNodes true(避免路径校验)
319.      * @return SessionDataSet
320.      */
321.     @Override
322.     public SessionDataSet executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes) {
323.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
324.         SessionDataSetWrapper sessionDataSetWrapper = null;
325.         try {
326.             log.info("iotdb数据查询：db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}]", db, device, sensors, isLegalPathNodes);
327.             sessionDataSetWrapper = sessionPool.executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
328.             return sessionDataSetWrapper.getSessionDataSet();
329.         } catch (Exception e) {
330.             log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
331.         }
332.         return null;
333.     }
335.     /**
336.      * @param db               root.ln.wf01
337.      * @param device           root.ln.wf01.wt01
338.      * @param sensors          temperature，status(字段名)
339.      * @param isLegalPathNodes true(避免路径校验)
340.      * @param clazz            返回数据对应的对象（对象属性必须与字段名对应）
341.      * @param <T>
342.      * @return
343.      */
344.     @Override
345.     public <T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes, Class<? extends IotDbRecordAble> clazz) {
346.         SessionDataSet sessionDataSet = executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
347.         List<String> columnNames = sessionDataSet.getColumnNames();
348.         List<T> resultEntities = null;
349.         try {
350.             resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
351.         } catch (Exception e) {
352.             log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
353.         }
354.         return resultEntities;
355.     }
357.     /**
358.      * 聚合查询
359.      *
360.      * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
361.      * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
362.      * @return SessionDataSet
363.      */
364.     @Override
365.     public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations) {
366.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
367.         SessionDataSetWrapper sessionDataSetWrapper = null;
368.         try {
369.             log.info("iotdb聚合查询：paths:[{}], aggregations：[{}]", paths, aggregations);
370.             sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations);
371.             return sessionDataSetWrapper.getSessionDataSet();
372.         } catch (Exception e) {
373.             log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,error={}", paths, aggregations, e.getMessage());
374.         }
375.         return null;
376.     }
378.     /**
379.      * 聚合查询(时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间)
380.      *
381.      * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
382.      * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
383.      * @param startTime    开始时间(包含)
384.      * @param endTime      结束时间
385.      * @return SessionDataSet
386.      */
387.     @Override
388.     public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime) {
389.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
390.         SessionDataSetWrapper sessionDataSetWrapper = null;
391.         try {
392.             log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}]", paths, aggregations, startTime, endTime);
393.             sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime);
394.             return sessionDataSetWrapper.getSessionDataSet();
395.         } catch (Exception e) {
396.             log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}],error={}", paths, aggregations, startTime, endTime, e.getMessage());
397.         }
398.         return null;
399.     }
401.     /**
402.      * 聚合查询(支持按照时间区间分段查询)
403.      *
404.      * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
405.      * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
406.      * @param startTime    开始时间(包含)
407.      * @param endTime      结束时间
408.      * @param interval
409.      * @return SessionDataSet
410.      */
411.     @Override
412.     public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval) {
413.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
414.         SessionDataSetWrapper sessionDataSetWrapper = null;
415.         try {
416.             log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}] ,interval:[{}]", paths, aggregations, startTime, endTime, interval);
417.             sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval);
418.             return sessionDataSetWrapper.getSessionDataSet();
419.         } catch (Exception e) {
420.             log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}], interval:[{}], error={}", paths, aggregations, startTime, endTime, interval, e.getMessage());
421.         }
422.         return null;
423.     }
425.     /**
426.      * 聚合查询(支持按照时间区间分段查询)
427.      *
428.      * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
429.      * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
430.      * @param startTime    开始时间(包含)
431.      * @param endTime      结束时间
432.      * @param interval
433.      * @param slidingStep
434.      * @return SessionDataSet
435.      */
436.     @Override
437.     public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval, long slidingStep) {
438.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
439.         SessionDataSetWrapper sessionDataSetWrapper = null;
441.         try {
442.             log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}] ,interval:[{}], slidingStep:[{}]", paths, aggregations, startTime, endTime, interval, slidingStep);
443.             sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval, slidingStep);
444.             return sessionDataSetWrapper.getSessionDataSet();
445.         } catch (Exception e) {
446.             log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}], interval:[{}], slidingStep:[{}] ,error={}", paths, aggregations, startTime, endTime, interval, slidingStep, e.getMessage());
447.         }
448.         return null;
449.     }
451.     /**
452.      * SQL查询
453.      *
454.      * @param sql
455.      * @return
456.      */
457.     @Override
458.     public SessionDataSet executeQueryStatement(String sql) {
459.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
460.         SessionDataSetWrapper sessionDataSetWrapper = null;
462.         try {
463.             log.info("iotdb SQL查询：sql:[{}]", sql);
464.              sessionDataSetWrapper = sessionPool.executeQueryStatement(sql);
465.             return sessionDataSetWrapper.getSessionDataSet();
466.         } catch (Exception e) {
467.             log.error("IotDBSession executeQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
468.         }
469.         return null;
470.     }
472.     /**
473.      * SQL非查询
474.      *
475.      * @param sql
476.      */
477.     @Override
478.     public void executeNonQueryStatement(String sql) {
479.         SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
480.         try {
481.             log.info("iotdb SQL无查询：sql:[{}]", sql);
482.             sessionPool.executeNonQueryStatement(sql);
483.         } catch (Exception e) {
484.             log.error("IotDBSession executeNonQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
485.         }
486.     }
488.     /**
489.      * 封装处理数据
490.      *
491.      * @param sessionDataSet
492.      * @param titleList
493.      */
494.     @SneakyThrows
495.     @Override
496.     public List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> titleList) {
497.         int fetchSize = sessionDataSet.getFetchSize();
498.         List<Map<String, Object>> resultList = new ArrayList<>();
499.         titleList.remove("Time");
500.         if (fetchSize > 0) {
501.             while (sessionDataSet.hasNext()) {
502.                 Map<String, Object> resultMap = new HashMap<>();
503.                 RowRecord next = sessionDataSet.next();
504.                 List<Field> fields = next.getFields();
505.                 String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
506.                 resultMap.put("time", timeString);
507.                 for (int i = 0; i < fields.size(); i++) {
508.                     Field field = fields.get(i);
509.                     if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
510.                         resultMap.put(splitString(titleList.get(i)), null);
511.                     } else {
512.                         resultMap.put(splitString(titleList.get(i)), field.getObjectValue(field.getDataType()).toString());
513.                     }
514.                 }
515.                 resultList.add(resultMap);
516.             }
517.         }
518.         return resultList;
519.     }
521.     /**
522.      * 封装处理数据（不支持聚合查询）
523.      *
524.      * @param sessionDataSet 查询返回的结果集
525.      * @param titleList      查询返回的结果集内的字段名
526.      * @param clazz          返回数据对应的对象（对象属性必须与字段名对应）
527.      * @param <T>
528.      * @return
529.      */
530.     @SneakyThrows
531.     @Override
532.     public <T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList, Class<? extends IotDbRecordAble> clazz) {
533.         int fetchSize = sessionDataSet.getFetchSize();
534.         List<T> resultList = new ArrayList<>();
535.         titleList.remove("Time");
536.         if (fetchSize > 0) {
537.             while (sessionDataSet.hasNext()) {
538.                 Map<String, Object> resultMap = new HashMap<>();
539.                 RowRecord next = sessionDataSet.next();
540.                 List<Field> fields = next.getFields();
541.                 String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
542.                 resultMap.put("time", timeString);
543.                 if (titleList.stream().anyMatch(str -> str.contains("."))) {
544.                     for (int i = 0; i < fields.size(); i++) {
545.                         Field field = fields.get(i);
546.                         String title = titleList.get(i);
547.                         if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
548.                             resultMap.put(splitString(title), null);
549.                         } else {
550.                             resultMap.put(splitString(title), field.getObjectValue(field.getDataType()).toString());
551.                         }
552.                     }
553.                 } else {
554.                     Field fieldName = fields.get(0);
555.                     Field fieldValue = fields.get(1);
556.                     Field fieldDataType = fields.get(2);
557.                     if (fieldName.getDataType() != null && fieldName.getObjectValue(fieldName.getDataType()) != null) {
558.                         String mapKey = fieldName.getObjectValue(fieldName.getDataType()).toString();
559.                         Object mapValue = convertStringToType(fieldValue.getObjectValue(fieldValue.getDataType()).toString(), fieldDataType.getObjectValue(fieldDataType.getDataType()).toString());
560.                         resultMap.put(splitString(mapKey), mapValue);
561.                     }
562.                 }
564.                 String jsonString = JSON.toJSONString(resultMap);
565.                 resultList.add(JSON.parseObject(jsonString, (Type) clazz));
566.             }
567.         }
568.         return resultList;
569.     }
571.     /**
572.      * 分割获取字段名
573.      *
574.      * @param str
575.      * @return 字段名
576.      */
577.     public static String splitString(String str) {
578.         String[] parts = str.split("\\.");
579.         if (parts.length <= 0) {
580.             return str;
581.         } else {
582.             return parts[parts.length - 1];
583.         }
584.     }
586.     /**
587.      * 根据数据值和数据类型返回对应数据类型数据
588.      *
589.      * @param value    数据值
590.      * @param typeName 数据类型
591.      * @return 转换后的数据值
592.      */
593.     public static Object convertStringToType(String value, String typeName) {
594.         String type = typeName.toLowerCase();
595.         if (type.isEmpty()) {
596.             return value;
597.         }
598.         if ("boolean".equals(type)) {
599.             return Boolean.parseBoolean(value);
600.         } else if ("double".equals(type)) {
601.             return Double.parseDouble(value);
602.         } else if ("int32".equals(type)) {
603.             return Integer.parseInt(value);
604.         } else if ("int64".equals(type)) {
605.             return Long.parseLong(value);
606.         } else if ("float".equals(type)) {
607.             return Float.parseFloat(value);
608.         } else if ("text".equals(type)) {
609.             return value;
610.         } else {
611.             return value;
612.         }
613.     }
615.     /**
616.      * 根据对象属性的数据类型返回对应的TSDataType
617.      *
618.      * @param type 属性的数据类型
619.      * @return TSDataType
620.      */
621.     public static TSDataType getTsDataTypeByString(String type) {
622.         String typeName = splitString(type).toLowerCase();
623.         if ("boolean".equals(typeName)) {
624.             return TSDataType.BOOLEAN;
625.         } else if ("double".equals(typeName)) {
626.             return TSDataType.DOUBLE;
627.         } else if ("int".equals(typeName) || "integer".equals(typeName)) {
628.             return TSDataType.INT32;
629.         } else if ("long".equals(typeName)) {
630.             return TSDataType.INT64;
631.         } else if ("float".equals(typeName)) {
632.             return TSDataType.FLOAT;
633.         } else if ("text".equals(typeName)) {
634.             return TSDataType.TEXT;
635.         } else if ("string".equals(typeName)) {
636.             return TSDataType.TEXT;
637.         } else {
638.             return TSDataType.UNKNOWN;
639.         }
640.     }
642.     /**
643.      * 根据对象构建MeasurementSchemas
644.      *
645.      * @param obj 对象
646.      * @return
647.      */
648.     @Override
649.     public List<MeasurementSchema> buildMeasurementSchemas(Object obj) {
650.         java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
651.         List<MeasurementSchema> schemaList = Arrays.stream(fields).map(field -> new MeasurementSchema(field.getName(), getTsDataTypeByString(field.getType().getName()))).collect(Collectors.toList());
652.         return schemaList;
653.     }
655.     /**
656.      * 根据对象构建MeasurementSchemaValuesDTO
657.      *
658.      * @param obj 对象
659.      * @return
660.      */
661.     @SneakyThrows
662.     @Override
663.     public MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object obj) {
664.         MeasurementSchemaValuesDTO measurementSchemaValuesDTO = new MeasurementSchemaValuesDTO();
665.         java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
666.         List<MeasurementSchema> schemaList = new ArrayList<>();
667.         List<Object> values = new ArrayList<>();
668.         List<Integer> valuesIsNullIndex = new ArrayList<>();
669.         int valueIndex = 0;
670.         for (java.lang.reflect.Field field : fields) {
671.             MeasurementSchema measurementSchema = new MeasurementSchema(field.getName(), getTsDataTypeByString(field.getType().getName()));
672.             schemaList.add(measurementSchema);
673.             Object value = field.get(obj);
674.             if (value == null) {
675.                 valuesIsNullIndex.add(valueIndex);
676.             }
677.             values.add(value);
678.             valueIndex++;
679.         }
680.         measurementSchemaValuesDTO.setSchemaList(schemaList);
681.         measurementSchemaValuesDTO.setValues(values);
682.         return measurementSchemaValuesDTO;
683.     }
684. }

复制代码

1.6、IotDbTest

1. package com.example;
3. import com.example.dto.InsertDataDTO;
4. import com.example.dto.MeasurementSchemaValuesDTO;
5. import com.example.dto.ResultEntity;
6. import com.example.dto.TestDataType;
7. import com.example.service.IotDbService;
8. import lombok.SneakyThrows;
9. import org.apache.iotdb.common.rpc.thrift.TAggregationType;
10. import org.apache.iotdb.isession.SessionDataSet;
11. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
12. import org.apache.iotdb.tsfile.write.record.Tablet;
13. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
14. import org.junit.Test;
15. import org.junit.runner.RunWith;
16. import org.springframework.beans.factory.annotation.Autowired;
17. import org.springframework.boot.test.context.SpringBootTest;
18. import org.springframework.test.context.junit4.SpringRunner;
20. import java.util.ArrayList;
21. import java.util.List;
22. import java.util.Map;
23. import java.util.Random;
24. import java.util.stream.Collectors;
25. import java.util.stream.IntStream;
28. @SpringBootTest
29. @RunWith(SpringRunner.class)
30. public class IotDbTest {
32.     @Autowired
33.     private IotDbService iotDbService;
35.     private static final String ROOT_LN_WF01 = "root.ln.wf01";
36.     private static final String ROOT_LN_WF01_WT01 = "root.ln.wf01.wt01";
37.     private static final String ROOT_LN_WF01_WT01_STATUS = "root.ln.wf01.wt01.status";
38.     private static final String ROOT_LN_WF01_WT01_HARDWARE = "root.ln.wf01.wt01.hardware";
39.     private static final String ROOT_LN_WF01_WT01_TEMPERATURE = "root.ln.wf01.wt01.temperature";
40.     private static final String SELECT_ROOT_LN_WF01_WT01 = "select * from root.ln.wf01.wt01";
42.     private static Random random = new Random();
44.     /**
45.      * 测试单个设备批量插入数据
46.      */
47.     @Test
48.     public void testInsertTablet() {
49.         List<InsertDataDTO> insertDataDTOS = new InsertDataDTO().buildList();
50.         List<MeasurementSchema> schemaList = iotDbService.buildMeasurementSchemas(insertDataDTOS.get(0));
51.         String deviceId = ROOT_LN_WF01_WT01;
52.         int maxRowNumber = insertDataDTOS.size();
53.         Tablet tablet = new Tablet(deviceId, schemaList, maxRowNumber);
54.         long timestamp = System.currentTimeMillis();
55.         for (int row = 0; row < maxRowNumber; row++) {
56.             int rowIndex = tablet.rowSize++;
57.             tablet.addTimestamp(rowIndex, timestamp);
58.             MeasurementSchemaValuesDTO measurementSchemaValuesDTO = iotDbService.buildMeasurementSchemasAndValues(insertDataDTOS.get(row));
59.             List<Object> iotValues = measurementSchemaValuesDTO.getValues();
60.             List<MeasurementSchema> schemaList2 = measurementSchemaValuesDTO.getSchemaList();
61.             List<String> iotMeasurements = schemaList2.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList());
62.             for (int i = 0; i < iotMeasurements.size(); i++) {
63.                 tablet.addValue(iotMeasurements.get(i), rowIndex, iotValues.get(i));
64.             }
65.             if (tablet.rowSize == tablet.getMaxRowNumber()) {
66.                 iotDbService.insertTablet(tablet);
67.                 tablet.reset();
68.             }
69.             timestamp++;
70.         }
71.     }
73.     /**
74.      * 测试单个设备批量插入数据(可插入空值)
75.      */
76.     @Test
77.     public void testInsertTabletWithNullValues() {
78.         List<InsertDataDTO> insertDataDTOS = new InsertDataDTO().buildList();
79.         List<MeasurementSchema> schemaList = iotDbService.buildMeasurementSchemas(insertDataDTOS.get(0));
80.         String deviceId = ROOT_LN_WF01_WT01;
81.         int maxRowNumber = insertDataDTOS.size();
82.         Tablet tablet = new Tablet(deviceId, schemaList, maxRowNumber);
83.         long timestamp = System.currentTimeMillis();
84.         tablet.initBitMaps();
85.         for (int row = 0; row < maxRowNumber; row++) {
86.             int rowIndex = tablet.rowSize++;
87.             tablet.addTimestamp(rowIndex, timestamp);
88.             MeasurementSchemaValuesDTO measurementSchemaValuesDTO = iotDbService.buildMeasurementSchemasAndValues(insertDataDTOS.get(row));
89.             List<Object> iotValues = measurementSchemaValuesDTO.getValues();
90.             List<MeasurementSchema> schemaList2 = measurementSchemaValuesDTO.getSchemaList();
91.             List<String> iotMeasurements = schemaList2.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList());
92.             List<Integer> valueIsNullIndex = measurementSchemaValuesDTO.getValueIsNullIndex();
93.             IntStream.range(0, iotMeasurements.size()).forEach(i -> tablet.addValue(iotMeasurements.get(i), rowIndex, iotValues.get(i)));
94.             if (valueIsNullIndex != null && !valueIsNullIndex.isEmpty()) {
95.                 for (Integer isNullIndex : valueIsNullIndex) {
96.                     tablet.bitMaps[isNullIndex].mark(row);
97.                 }
98.             }
99.             if (tablet.rowSize == tablet.getMaxRowNumber()) {
100.                 iotDbService.insertTablet(tablet);
101.                 tablet.reset();
102.             }
103.             timestamp++;
104.         }
105.     }
107.     /**
108.      * 测试单条数据插入
109.      */
110.     @Test
111.     public void testInsertRecord() {
112.         String deviceId = ROOT_LN_WF01_WT01;
113.         long currentTime = System.currentTimeMillis();
114.         InsertDataDTO insertDataDto = new InsertDataDTO().buildOne();
115.         MeasurementSchemaValuesDTO measurementSchemaValuesDTO = iotDbService.buildMeasurementSchemasAndValues(insertDataDto);
116.         List<Object> iotValues = measurementSchemaValuesDTO.getValues();
117.         List<MeasurementSchema> schemaList = measurementSchemaValuesDTO.getSchemaList();
118.         List<String> iotMeasurements = schemaList.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList());
119.         List<TSDataType> tsDataTypeList = schemaList.stream().map(MeasurementSchema::getType).collect(Collectors.toList());
120.         iotDbService.insertRecord(deviceId, currentTime, iotMeasurements, tsDataTypeList, iotValues);
121.     }
123.     /**
124.      * 测试单个设备多条数据插入
125.      */
126.     @Test
127.     public void testInsertRecordsOfOneDevice() {
128.         String deviceId = ROOT_LN_WF01_WT01;
129.         List<InsertDataDTO> insertDataDTOS = new InsertDataDTO().buildList();
130.         int recordNum = insertDataDTOS.size();
131.         List<Long> timeList = new ArrayList<>();
132.         List<List<String>> iotMeasurementsList = new ArrayList<>();
133.         List<List<TSDataType>> tsDataTypesList = new ArrayList<>();
134.         List<List<Object>> valuesList = new ArrayList<>();
135.         for (int i = 0; i < recordNum; i++) {
136.             long currentTime = System.currentTimeMillis();
137.             timeList.add(currentTime + i);
139.             MeasurementSchemaValuesDTO measurementSchemaValuesDTO = iotDbService.buildMeasurementSchemasAndValues(insertDataDTOS.get(i));
140.             List<Object> iotValues = measurementSchemaValuesDTO.getValues();
141.             List<MeasurementSchema> schemaList = measurementSchemaValuesDTO.getSchemaList();
142.             List<String> iotMeasurements = schemaList.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList());
143.             List<TSDataType> tsDataTypeList = schemaList.stream().map(MeasurementSchema::getType).collect(Collectors.toList());
145.             iotMeasurementsList.add(iotMeasurements);
146.             tsDataTypesList.add(tsDataTypeList);
147.             valuesList.add(iotValues);
148.         }
150.         iotDbService.insertRecordsOfOneDevice(deviceId, timeList, iotMeasurementsList, tsDataTypesList, valuesList);
151.     }
153.     /**
154.      * 测试时间查询
155.      */
156.     @SneakyThrows
157.     @Test
158.     public void testExecuteRawDataQuery() {
159.         List<String> paths = new ArrayList<>();
160.         paths.add(ROOT_LN_WF01_WT01_TEMPERATURE);
161.         paths.add(ROOT_LN_WF01_WT01_HARDWARE);
162.         long startTime = 1718959439344L;
163.         long endTime = System.currentTimeMillis();
164.         long outTime = 60000;
165.         SessionDataSet sessionDataSet = iotDbService.executeRawDataQuery(paths, startTime, endTime, outTime);
166.         System.out.println(sessionDataSet.getColumnNames());
167.         sessionDataSet.setFetchSize(1024); // default is 10000
168.         while (sessionDataSet.hasNext()) {
169.             System.out.println(sessionDataSet.next());
170.         }
172.         List<ResultEntity> resultEntities = iotDbService.executeRawDataQuery(paths, startTime, endTime, outTime, ResultEntity.class);
173.         System.out.println(resultEntities);
174.         System.out.println(resultEntities.size());
175.     }
177.     /**
178.      * 测试单个设备最新时间点查询
179.      */
180.     @SneakyThrows
181.     @Test
182.     public void testExecuteLastDataQueryForOneDevice() {
183.         String db = ROOT_LN_WF01;
184.         String device = ROOT_LN_WF01_WT01;
185.         List<String> sensors = new ArrayList<>();
186.         sensors.add("temperature");
187.         sensors.add("hardware");
188.         sensors.add("status");
190.         SessionDataSet sessionDataSet = iotDbService.executeLastDataQueryForOneDevice(db, device, sensors, true);
191.         System.out.println(sessionDataSet.getColumnNames());
192.         sessionDataSet.setFetchSize(1024); // default is 10000
193.         while (sessionDataSet.hasNext()) {
194.             System.out.println(sessionDataSet.next());
195.         }
197.         List<ResultEntity> resultEntities = iotDbService.executeLastDataQueryForOneDevice(db, device, sensors, true, ResultEntity.class);
198.         System.out.println(resultEntities);
199.         System.out.println(resultEntities.size());
200.     }
202.     /**
203.      * 测试最新点查询
204.      */
205.     @SneakyThrows
206.     @Test
207.     public void testExecuteLastDataQuery() {
208.         List<String> paths = new ArrayList<>();
209.         paths.add(ROOT_LN_WF01_WT01_TEMPERATURE);
210.         paths.add(ROOT_LN_WF01_WT01_HARDWARE);
211.         long lastTime = 1718959439344L;
212.         long endTime = System.currentTimeMillis();
214.         SessionDataSet sessionDataSet = iotDbService.executeLastDataQuery(paths, lastTime);
215.         System.out.println(sessionDataSet.getColumnNames());
216.         sessionDataSet.setFetchSize(1024); // default is 10000
217.         while (sessionDataSet.hasNext()) {
218.             System.out.println(sessionDataSet.next());
219.         }
221.         List<ResultEntity> resultEntities = iotDbService.executeLastDataQuery(paths, lastTime, ResultEntity.class);
222.         System.out.println(resultEntities);
223.         System.out.println(resultEntities.size());
224.     }
226.     /**
227.      * 测试聚合查询
228.      */
229.     @SneakyThrows
230.     @Test
231.     public void testExecuteAggregationQuery() {
232.         List<String> paths = new ArrayList<>();
233.         paths.add(ROOT_LN_WF01_WT01_TEMPERATURE);
235.         List<TAggregationType> aggregations = new ArrayList<>();
236.         aggregations.add(TAggregationType.SUM);
238.         SessionDataSet sessionDataSet = iotDbService.executeAggregationQuery(paths, aggregations);
239.         System.out.println(sessionDataSet.getColumnNames());
240.         sessionDataSet.setFetchSize(1024); // default is 10000
241.         while (sessionDataSet.hasNext()) {
242.             System.out.println(sessionDataSet.next());
243.         }
244.     }
246.     @SneakyThrows
247.     @Test
248.     public void testQueryByIterator() {
249.         String sql = SELECT_ROOT_LN_WF01_WT01;
250.         SessionDataSet sessionDataSet = iotDbService.executeQueryStatement(sql);
251.         SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
252.         System.out.println(sessionDataSet.getColumnNames());
253.         sessionDataSet.setFetchSize(1024); // default is 10000
254.         while (iterator.next()) {
255.             float aFloat = iterator.getFloat(ROOT_LN_WF01_WT01_TEMPERATURE);
256.             String aString = iterator.getString(ROOT_LN_WF01_WT01_HARDWARE);
257.             boolean aBoolean = iterator.getBoolean(ROOT_LN_WF01_WT01_STATUS);
258.             System.out.println(aFloat);
259.             System.out.println(aString);
260.             System.out.println(aBoolean);
261.         }
262.     }
264.     /**
265.      * 测试删除数据
266.      */
267.     @Test
268.     public void testDeleteData() {
269.         List<String> paths = new ArrayList<>();
270.         paths.add(ROOT_LN_WF01_WT01_TEMPERATURE);
271.         paths.add(ROOT_LN_WF01_WT01_STATUS);
272.         paths.add(ROOT_LN_WF01_WT01_HARDWARE);
273.         long endTime = 1719212858916L;
274.         iotDbService.deleteData(paths, endTime);
275.     }
277.     /**
278.      * 测试SQL查询
279.      */
280.     @SneakyThrows
281.     @Test
282.     public void testExecuteQueryStatement() {
283.         String sql = SELECT_ROOT_LN_WF01_WT01;
284.         SessionDataSet sessionDataSet = iotDbService.executeQueryStatement(sql);
285.         System.out.println(sessionDataSet.getColumnNames());
286.         sessionDataSet.setFetchSize(1024); // default is 10000
287.         while (sessionDataSet.hasNext()) {
288.             System.out.println(sessionDataSet.next()    );
289.         }
290.     }
292.     /**
293.      * 测试SQL执行（不包含查询）
294.      */
295.     @Test
296.     public void testExecuteNonQueryStatement() {
297.         String sql = "insert into root.ln.wf01.wt01(temperature, hardware, status) values (2.1, 'v2', false)";
298.         iotDbService.executeNonQueryStatement(sql);
299.     }
301.     /**
302.      * 测试封装查询结果
303.      */
304.     @Test
305.     public void testPackagingMapData() {
306.         String sql = SELECT_ROOT_LN_WF01_WT01;
307.         SessionDataSet sessionDataSet = iotDbService.executeQueryStatement(sql);
308.         List<String> columnNames = sessionDataSet.getColumnNames();
309.         List<Map<String, Object>> rtList = iotDbService.packagingMapData(sessionDataSet, columnNames);
310.         System.out.println(rtList);
311.     }
313.     /**
314.      * 测试封装查询结果
315.      */
316.     @Test
317.     public void testPackagingObjectData() {
318.         String sql = SELECT_ROOT_LN_WF01_WT01;
319.         SessionDataSet sessionDataSet = iotDbService.executeQueryStatement(sql);
320.         List<String> columnNames = sessionDataSet.getColumnNames();
321.         List<ResultEntity> resultEntities = iotDbService.packagingObjectData(sessionDataSet, columnNames, ResultEntity.class);
322.         System.out.println(resultEntities);
323.     }
325.     /**
326.      * 测试构建MeasurementSchemasAndValues
327.      */
328.     @Test
329.     public void testBuildMeasurementSchema() {
330.         TestDataType testDataType = new TestDataType();
331.         testDataType.setTestDouble(12D);
332.         testDataType.setTestLong(12L);
333.         testDataType.setHardware("ss");
334.         testDataType.setStatus(true);
335.         testDataType.setTemperature(12.0F);
337.         List<MeasurementSchema> schemaList = iotDbService.buildMeasurementSchemas(testDataType);
338.         System.out.println(schemaList);
339.         List<String> iotMeasurements = schemaList.stream().map(MeasurementSchema::getMeasurementId).collect(Collectors.toList());
340.         System.out.println(iotMeasurements);
341.         List<TSDataType> tsDataTypeList = schemaList.stream().map(MeasurementSchema::getType).collect(Collectors.toList());
342.         System.out.println(tsDataTypeList);
343.         MeasurementSchemaValuesDTO measurementSchemaValuesDTO = iotDbService.buildMeasurementSchemasAndValues(testDataType);
344.         System.out.println(measurementSchemaValuesDTO.getSchemaList());
345.         System.out.println(measurementSchemaValuesDTO.getValues());
347.     }
348. }

复制代码

1.7、Entity & DTO

1. package com.example.dto;
3. import lombok.Data;
5. import java.util.ArrayList;
6. import java.util.List;
8. @Data
9. public class InsertDataDTO {
10.     public Float temperature;
11.     public String hardware;
12.     public Boolean status;
14.     public InsertDataDTO buildOne() {
15.         InsertDataDTO insertDataDTO = new InsertDataDTO();
16.         insertDataDTO.setHardware("ss");
17.         insertDataDTO.setStatus(true);
18.         insertDataDTO.setTemperature(12.0F);
19.         return insertDataDTO;
20.     }
22.     public List<InsertDataDTO> buildList() {
23.         List<InsertDataDTO> insertDataDTOS = new ArrayList<>();
24.         int buildNum = 10;
25.         for (int i = 0; i < buildNum; i++) {
26.             InsertDataDTO insertDataDTO = new InsertDataDTO();
27.             insertDataDTO.setHardware(i % 2 == 0 ? "pp" + i : null);
28.             insertDataDTO.setStatus(i % 2 == 0);
29.             insertDataDTO.setTemperature(12.0F + i);
30.             insertDataDTOS.add(insertDataDTO);
31.         }
32.         return insertDataDTOS;
33.     }
34. }

复制代码

1. package com.example.dto;
3. import lombok.Data;
5. @Data
6. public class IotDbRecordAble {
7. }

复制代码

1. package com.example.dto;
3. import lombok.Data;
4. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
6. import java.util.List;
9. @Data
10. public class MeasurementSchemaValuesDTO {
11.     List<MeasurementSchema> schemaList;
12.     List<Object> values;
13.     List<Integer> valueIsNullIndex;
14. }

复制代码

1. package com.example.dto;
3. import lombok.Data;
5. @Data
6. public class ResultEntity extends IotDbRecordAble {
7.     public Float temperature;
8.     public String hardware;
9.     public Boolean status;
10.     public String time;
11. }

复制代码

1. package com.example.dto;
3. import lombok.Data;
5. @Data
6. public class TestDataType {
7.     public Float temperature;
8.     public String hardware;
9.     public Boolean status;
10.     public Double testDouble;
11.     public Long testLong;
12. }

复制代码

这里利用的是Java原生接口的方式来访问数据库，进行新增、查询、删除，适用于高吞吐性能场景，大批量的数据处理。下一节内容采用mybatis的方式。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。