From 9da2194aa665e4f960aa4171db617e0b17e52086 Mon Sep 17 00:00:00 2001 From: xiaohuang <1559741705@qq.com> Date: Thu, 23 May 2024 19:37:03 +0800 Subject: [PATCH] lotdb --- muyu-modules/muyu-data-iotdb/pom.xml | 66 +++++++ .../com/muyu/iotdb/MuyuIotdbApplication.java | 19 ++ .../muyu/iotdb/config/IotDBSessionConfig.java | 187 ++++++++++++++++++ .../iotdb/controller/IotDbController.java | 69 +++++++ .../com/muyu/iotdb/domain/IotDbParam.java | 43 ++++ .../com/muyu/iotdb/domain/IotDbResult.java | 36 ++++ .../com/muyu/iotdb/service/IotDbServer.java | 17 ++ .../iotdb/service/impl/IotDbServerImpl.java | 132 +++++++++++++ muyu-modules/muyu-system/pom.xml | 40 ++++ muyu-modules/pom.xml | 1 + pom.xml | 1 + 11 files changed, 611 insertions(+) create mode 100644 muyu-modules/muyu-data-iotdb/pom.xml create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/MuyuIotdbApplication.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/config/IotDBSessionConfig.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/controller/IotDbController.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbParam.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbResult.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/IotDbServer.java create mode 100644 muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/impl/IotDbServerImpl.java diff --git a/muyu-modules/muyu-data-iotdb/pom.xml b/muyu-modules/muyu-data-iotdb/pom.xml new file mode 100644 index 0000000..2b36b69 --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/pom.xml @@ -0,0 +1,66 @@ + + + 4.0.0 + + com.muyu + muyu-modules + 3.6.3 + + + muyu-data-iotdb + + + 17 + 17 + UTF-8 + + + + org.apache.iotdb + iotdb-session + 0.14.0-preview1 + + + + cn.hutool + hutool-all + 5.6.3 + + + + com.alibaba + fastjson + 1.2.83 + + + com.bwie + bwie-common + 1.0-SNAPSHOT + + + + org.springframework.boot + spring-boot-starter-web + + + + org.projectlombok + lombok + true + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + + diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/MuyuIotdbApplication.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/MuyuIotdbApplication.java new file mode 100644 index 0000000..1107558 --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/MuyuIotdbApplication.java @@ -0,0 +1,19 @@ +package com.muyu.iotdb; + + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName MuyuIotdbApplication + * @Date 2024/05/21 17:01 + */ +@SpringBootApplication +public class MuyuIotdbApplication { + public static void main(String[] args) { + SpringApplication.run(MuyuIotdbApplication.class); + } +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/config/IotDBSessionConfig.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/config/IotDBSessionConfig.java new file mode 100644 index 0000000..7aabdaf --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/config/IotDBSessionConfig.java @@ -0,0 +1,187 @@ +package com.muyu.iotdb.config; + +import lombok.extern.log4j.Log4j2; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; +import org.apache.iotdb.session.SessionDataSet; +import org.apache.iotdb.session.util.Version; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.write.record.Tablet; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.stereotype.Component; + +import java.rmi.ServerException; +import java.util.ArrayList; +import java.util.List; + + +/** + * description: iotdb 配置工具类（常用部分，如需要可以自行扩展） + * 注意：可以不需要创建分组，插入时默认前两个节点名称为分组名称 比如： root.a1eaKSRpRty.CA3013A303A25467 或者 + * root.a1eaKSRpRty.CA3013A303A25467.heart 他们的分组都为 root.a1eaKSRpRty + * author: zhouhong + */ +@Log4j2 +@Component +@Configuration +public class IotDBSessionConfig { + + private static Session session; + private static final String LOCAL_HOST = "175.24.138.82"; + @Bean + public Session getSession() throws IoTDBConnectionException, StatementExecutionException { + if (session == null) { + log.info("正在连接iotdb......."); + session = new Session.Builder().host(LOCAL_HOST).port(6667).username("root").password("root").version(Version.V_0_13).build(); + session.open(false); + session.setFetchSize(100); + log.info("iotdb连接成功~"); + // 设置时区 + session.setTimeZone("+08:00"); + } + return session; + } + + /** + * description: 带有数据类型的添加操作 - insertRecord没有指定类型 + * author: zhouhong + * @param * @param deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467 + * time:时间戳 + * measurementsList：物理量 即：属性 + * type：数据类型： BOOLEAN((byte)0), INT32((byte)1),INT64((byte)2),FLOAT((byte)3),DOUBLE((byte)4),TEXT((byte)5),VECTOR((byte)6); + * valuesList：属性值 --- 属性必须与属性值一一对应 + * @return + */ + public void insertRecordType(String deviceId, Long time,List measurementsList, TSDataType type,List valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException { + if (measurementsList.size() != valuesList.size()) { + throw new ServerException("measurementsList 与 valuesList 值不对应"); + } + List types = new ArrayList<>(); + measurementsList.forEach(item -> { + types.add(type); + }); + session.insertRecord(deviceId, time, measurementsList, types, valuesList); + } + /** + * description: 带有数据类型的添加操作 - insertRecord没有指定类型 + * author: zhouhong + * @param deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467 + * @param time:时间戳 + * @param measurementsList：物理量 即：属性 + * @param valuesList：属性值 --- 属性必须与属性值一一对应 + * @return + */ + public void insertRecord(String deviceId, Long time,List measurementsList, List valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException { + if (measurementsList.size() == valuesList.size()) { + session.insertRecord(deviceId, time, measurementsList, valuesList); + } else { + log.error("measurementsList 与 valuesList 值不对应"); + } + } + /** + * description: 批量插入 + * author: zhouhong + */ + public void insertRecords(List deviceIdList, List timeList, List> measurementsList, List> valuesList) throws StatementExecutionException, IoTDBConnectionException, ServerException { + if (measurementsList.size() == valuesList.size()) { + session.insertRecords(deviceIdList, timeList, measurementsList, valuesList); + } else { + log.error("measurementsList 与 valuesList 值不对应"); + } + } + + /** + * description: 插入操作 + * author: zhouhong + * @param deviceId:节点路径如：root.a1eaKSRpRty.CA3013A303A25467 + * @param time:时间戳 + * @param schemaList: 属性值 + 数据类型 例子： List schemaList = new ArrayList<>(); schemaList.add(new MeasurementSchema("breath", TSDataType.INT64)); + * @param maxRowNumber： + * @return + */ + public void insertTablet(String deviceId, Long time,List schemaList, List valueList,int maxRowNumber) throws StatementExecutionException, IoTDBConnectionException { + + Tablet tablet = new Tablet(deviceId, schemaList, maxRowNumber); + // 向iotdb里面添加数据 + int rowIndex = tablet.rowSize++; + tablet.addTimestamp(rowIndex, time); + for (int i = 0; i < valueList.size(); i++) { + tablet.addValue(schemaList.get(i).getMeasurementId(), rowIndex, valueList.get(i)); + } + if (tablet.rowSize == tablet.getMaxRowNumber()) { + session.insertTablet(tablet, true); + tablet.reset(); + } + if (tablet.rowSize != 0) { + session.insertTablet(tablet); + tablet.reset(); + } + } + + /** + * description: 根据SQL查询 + * author: zhouhong + */ + public SessionDataSet query(String sql) throws StatementExecutionException, IoTDBConnectionException { + return session.executeQueryStatement(sql); + } + + /** + * description: 删除分组 如 root.a1eaKSRpRty + * author: zhouhong + * @param groupName：分组名称 + * @return + */ + public void deleteStorageGroup(String groupName) throws StatementExecutionException, IoTDBConnectionException { + session.deleteStorageGroup(groupName); + } + + /** + * description: 根据Timeseries删除 如：root.a1eaKSRpRty.CA3013A303A25467.breath （个人理解：为具体的物理量） + * author: zhouhong + */ + public void deleteTimeseries(String timeseries) throws StatementExecutionException, IoTDBConnectionException { + session.deleteTimeseries(timeseries); + } + /** + * description: 根据Timeseries批量删除 + * author: zhouhong + */ + public void deleteTimeserieList(List timeseriesList) throws StatementExecutionException, IoTDBConnectionException { + session.deleteTimeseries(timeseriesList); + } + + /** + * description: 根据分组批量删除 + * author: zhouhong + */ + public void deleteStorageGroupList(List storageGroupList) throws StatementExecutionException, IoTDBConnectionException { + session.deleteStorageGroups(storageGroupList); + } + + /** + * description: 根据路径和结束时间删除 结束时间之前的所有数据 + * author: zhouhong + */ + public void deleteDataByPathAndEndTime(String path, Long endTime) throws StatementExecutionException, IoTDBConnectionException { + session.deleteData(path, endTime); + } + /** + * description: 根据路径集合和结束时间批量删除 结束时间之前的所有数据 + * author: zhouhong + */ + public void deleteDataByPathListAndEndTime(List pathList, Long endTime) throws StatementExecutionException, IoTDBConnectionException { + session.deleteData(pathList, endTime); + } + /** + * description: 根据路径集合和时间段批量删除 + * author: zhouhong + */ + public void deleteDataByPathListAndTime(List pathList, Long startTime,Long endTime) throws StatementExecutionException, IoTDBConnectionException { + session.deleteData(pathList, startTime, endTime); + } + +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/controller/IotDbController.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/controller/IotDbController.java new file mode 100644 index 0000000..5bfb07f --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/controller/IotDbController.java @@ -0,0 +1,69 @@ +package com.muyu.iotdb.controller; + + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName IotDbController + * @Date 2024/05/21 16:55 + */ + +import com.bwie.common.result.Result; +import com.muyu.iotdb.config.IotDBSessionConfig; +import com.muyu.iotdb.domain.IotDbParam; +import com.muyu.iotdb.service.IotDbServer; +import lombok.extern.log4j.Log4j2; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +import javax.annotation.Resource; +import java.rmi.ServerException; + +/** + * description: iotdb 控制层 + * date: 2022/8/15 21:50 + * author: zhouhong + */ +@Log4j2 +@RestController +public class IotDbController { + + @Resource + private IotDbServer iotDbServer; + @Resource + private IotDBSessionConfig iotDBSessionConfig; + + /** + * 插入数据 + * @param iotDbParam + */ + @PostMapping("/api/device/insert") + public Result insert(@RequestBody IotDbParam iotDbParam) throws StatementExecutionException, ServerException, IoTDBConnectionException { + iotDbServer.insertData(iotDbParam); + return Result.success(); + } + + /** + * 插入数据 + * @param iotDbParam + */ + @PostMapping("/api/device/queryData") + public Result queryDataFromIotDb(@RequestBody IotDbParam iotDbParam) throws Exception { + return Result.success(iotDbServer.queryDataFromIotDb(iotDbParam)); + } + /** + * 删除分组 + * @return + */ + @PostMapping("/api/device/deleteGroup") + public Result deleteGroup() throws StatementExecutionException, IoTDBConnectionException { + iotDBSessionConfig.deleteStorageGroup("root.a1eaKSRpRty"); + iotDBSessionConfig.deleteStorageGroup("root.smartretirement"); + return Result.success(); + } + +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbParam.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbParam.java new file mode 100644 index 0000000..07ada7e --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbParam.java @@ -0,0 +1,43 @@ +package com.muyu.iotdb.domain; + + +import lombok.Data; + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName IotDbParam + * @Date 2024/05/21 16:06 + */ +@Data +public class IotDbParam { + /** + * 产品PK + */ + private String pk; + /** + * 设备号 + */ + private String sn; + /** + * 时间 + */ + private Long time; + /** + * 实时呼吸 + */ + private String breath; + /** + * 实时心率 + */ + private String heart; + /** + * 查询开始时间 + */ + private String startTime; + /** + * 查询结束时间 + */ + private String endTime; +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbResult.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbResult.java new file mode 100644 index 0000000..54eac54 --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/domain/IotDbResult.java @@ -0,0 +1,36 @@ +package com.muyu.iotdb.domain; + + +import lombok.Data; + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName IotDbResult + * @Date 2024/05/21 16:10 + */ +@Data +public class IotDbResult { + /** + * 时间 + */ + private String time; + /** + * 产品PK + */ + private String pk; + /** + * 设备号 + */ + private String sn; + /** + * 实时呼吸 + */ + private String breath; + /** + * 实时心率 + */ + private String heart; + +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/IotDbServer.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/IotDbServer.java new file mode 100644 index 0000000..b859602 --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/IotDbServer.java @@ -0,0 +1,17 @@ +package com.muyu.iotdb.service; + + +import com.muyu.iotdb.domain.IotDbParam; + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName IotDbServer + * @Date 2024/05/21 16:56 + */ +public interface IotDbServer { + void insertData(IotDbParam iotDbParam); + + Object queryDataFromIotDb(IotDbParam iotDbParam); +} diff --git a/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/impl/IotDbServerImpl.java b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/impl/IotDbServerImpl.java new file mode 100644 index 0000000..ee2de7c --- /dev/null +++ b/muyu-modules/muyu-data-iotdb/src/main/java/com/muyu/iotdb/service/impl/IotDbServerImpl.java @@ -0,0 +1,132 @@ +package com.muyu.iotdb.service.impl; + + +import com.muyu.iotdb.config.IotDBSessionConfig; +import com.muyu.iotdb.domain.IotDbParam; +import com.muyu.iotdb.domain.IotDbResult; +import com.muyu.iotdb.service.IotDbServer; +import lombok.extern.log4j.Log4j2; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.SessionDataSet; +import org.apache.iotdb.tsfile.read.common.Field; +import org.apache.iotdb.tsfile.read.common.RowRecord; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.rmi.ServerException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * BingRui.Hou + * + * @Description 描述 + * @ClassName IotDbServerImpl + * @Date 2024/05/21 16:55 + */ +@Log4j2 +@Service +public class IotDbServerImpl implements IotDbServer { + + @Resource + private IotDBSessionConfig iotDBSessionConfig; + + @Override + public void insertData(IotDbParam iotDbParam) { + // iotDbParam: 模拟设备上报消息 + // bizkey: 业务唯一key PK :产品唯一编码 SN:设备唯一编码 + String deviceId = "root.bizkey."+ iotDbParam.getPk() + "." + iotDbParam.getSn(); + // 将设备上报的数据存入数据库(时序数据库) + List measurementsList = new ArrayList<>(); + measurementsList.add("heart"); + measurementsList.add("breath"); + List valuesList = new ArrayList<>(); + valuesList.add(String.valueOf(iotDbParam.getHeart())); + valuesList.add(String.valueOf(iotDbParam.getBreath())); + try { + iotDBSessionConfig.insertRecord(deviceId, iotDbParam.getTime(), measurementsList, valuesList); + } catch (IoTDBConnectionException e) { + throw new RuntimeException(e); + } catch (StatementExecutionException e) { + throw new RuntimeException(e); + } catch (ServerException e) { + throw new RuntimeException(e); + } + } + + @Override + public List queryDataFromIotDb(IotDbParam iotDbParam) { + List iotDbResultList = new ArrayList<>(); + + if (null != iotDbParam.getPk() && null != iotDbParam.getSn()) { + String sql = "select * from root.bizkey."+ iotDbParam.getPk() +"." + iotDbParam.getSn() + " where time >= " + + iotDbParam.getStartTime() + " and time < " + iotDbParam.getEndTime(); + SessionDataSet sessionDataSet = null; + try { + sessionDataSet = iotDBSessionConfig.query(sql); + } catch (StatementExecutionException e) { + throw new RuntimeException(e); + } catch (IoTDBConnectionException e) { + throw new RuntimeException(e); + } + List columnNames = sessionDataSet.getColumnNames(); + List titleList = new ArrayList<>(); + // 排除Time字段 -- 方便后面后面拼装数据 + for (int i = 1; i < columnNames.size(); i++) { + String[] temp = columnNames.get(i).split("\\."); + titleList.add(temp[temp.length - 1]); + } + // 封装处理数据 + try { + packagingData(iotDbParam, iotDbResultList, sessionDataSet, titleList); + } catch (StatementExecutionException e) { + throw new RuntimeException(e); + } catch (IoTDBConnectionException e) { + throw new RuntimeException(e); + } + } else { + log.info("PK或者SN不能为空！！"); + } + return iotDbResultList; + } + /** + * 封装处理数据 + * @param iotDbParam + * @param iotDbResultList + * @param sessionDataSet + * @param titleList + * @throws StatementExecutionException + * @throws IoTDBConnectionException + */ + private void packagingData(IotDbParam iotDbParam, List iotDbResultList, SessionDataSet sessionDataSet, List titleList) + throws StatementExecutionException, IoTDBConnectionException { + int fetchSize = sessionDataSet.getFetchSize(); + if (fetchSize > 0) { + while (sessionDataSet.hasNext()) { + IotDbResult iotDbResult = new IotDbResult(); + RowRecord next = sessionDataSet.next(); + List fields = next.getFields(); + String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp()); + iotDbResult.setTime(timeString); + Map map = new HashMap<>(); + + for (int i = 0; i < fields.size(); i++) { + Field field = fields.get(i); + // 这里的需要按照类型获取 + map.put(titleList.get(i), field.getObjectValue(field.getDataType()).toString()); + } + iotDbResult.setTime(timeString); + iotDbResult.setPk(iotDbParam.getPk()); + iotDbResult.setSn(iotDbParam.getSn()); + iotDbResult.setHeart(map.get("heart")); + iotDbResult.setBreath(map.get("breath")); + iotDbResultList.add(iotDbResult); + } + } + } + + +} diff --git a/muyu-modules/muyu-system/pom.xml b/muyu-modules/muyu-system/pom.xml index be58498..cc79234 100644 --- a/muyu-modules/muyu-system/pom.xml +++ b/muyu-modules/muyu-system/pom.xml @@ -78,6 +78,46 @@ muyu-common-swagger + + + org.apache.iotdb + iotdb-session + 0.14.0-preview1 + + + + cn.hutool + hutool-all + 5.6.3 + + + + com.alibaba + fastjson + 1.2.83 + + + + org.springframework.boot + spring-boot-starter-web + + + + org.projectlombok + lombok + true + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + diff --git a/muyu-modules/pom.xml b/muyu-modules/pom.xml index 0e34d6c..7ef2e65 100644 --- a/muyu-modules/pom.xml +++ b/muyu-modules/pom.xml @@ -15,6 +15,7 @@ muyu-file muyu-etl muyu-rule-engine + muyu-data-iotdb diff --git a/pom.xml b/pom.xml index c7d7501..ba3a254 100644 --- a/pom.xml +++ b/pom.xml @@ -215,6 +215,7 @@ muyu-visual muyu-modules muyu-common + muyu-modules/muyu-data-iotdb pom