tdengine数据库支持

pnoker · Jun 12, 2023 · 00c2b4d · 00c2b4d
1 parent b015dc7
commit 00c2b4d
Show file tree

Hide file tree

Showing 5 changed files with 107 additions and 2 deletions.
diff --git a/dc3-center/dc3-center-data/pom.xml b/dc3-center/dc3-center-data/pom.xml
@@ -84,6 +84,13 @@
             <version>2023.4.3</version>
         </dependency>
 
+        <!-- DC3 Common tdengine -->
+        <dependency>
+            <groupId>io.github.pnoker</groupId>
+            <artifactId>dc3-common-tdengine</artifactId>
+            <version>2023.4.3</version>
+        </dependency>
+
         <!-- DC3 Common Quartz -->
         <dependency>
             <groupId>io.github.pnoker</groupId>

diff --git a/...3-center-data/src/main/java/io/github/pnoker/center/data/mapper/TaosPointValueMapper.java b/...3-center-data/src/main/java/io/github/pnoker/center/data/mapper/TaosPointValueMapper.java
@@ -0,0 +1,28 @@
+package io.github.pnoker.center.data.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import io.github.pnoker.common.entity.point.TaosPointValue;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
+
+import java.util.List;
+
+@Mapper
+@DS("taos")
+public interface TaosPointValueMapper extends BaseMapper<TaosPointValue> {
+
+    @Update("CREATE STABLE IF NOT EXISTS point_value (create_time TIMESTAMP, point_value NCHAR(32), raw_value NCHAR(32),  origin_time TIMESTAMP) TAGS (device_id NCHAR(32), point_id NCHAR(32))")
+    int createSuperTable();
+
+    @Update("CREATE TABLE IF NOT EXISTS point_value_${deviceId} using point_value TAGS (#{deviceId},#{pointId})")
+    int createDeviceTable(@Param("deviceId") String deviceId, @Param("pointId") String pointId);
+
+    @Insert("INSERT INTO point_value_${deviceId} (create_time,point_value,raw_value,origin_time) VALUES (#{createTime},#{pointValue},#{rawValue},#{originTime})")
+    int insertOne(TaosPointValue taosPointValue);
+
+    @Insert("<script>INSERT INTO point_value_${deviceId} (create_time,point_value,raw_value,origin_time) VALUES <foreach collection='list' item='item' index='index' separator=','>(#{item.createTime},#{item.pointValue},#{item.rawValue},#{item.originTime})</foreach></script>")
+    int batchInsert(List<TaosPointValue> collect);
+}
diff --git a/.../src/main/java/io/github/pnoker/center/data/service/impl/RepositoryHandleServiceImpl.java b/.../src/main/java/io/github/pnoker/center/data/service/impl/RepositoryHandleServiceImpl.java
@@ -41,6 +41,8 @@ public class RepositoryHandleServiceImpl implements RepositoryHandleService {
 
     @Value("${data.point.sava.influxdb.enable}")
     private Boolean enableInfluxdb;
+    @Value("${data.point.sava.tdengine.enable}")
+    private Boolean enableTDengine;
     @Value("${data.point.sava.opentsdb.enable}")
     private Boolean enableOpentsdb;
     @Value("${data.point.sava.elasticsearch.enable}")
@@ -61,11 +63,16 @@ public void save(PointValue pointValue) {
 
         // 保存单个数据到 Influxdb
         if (Boolean.TRUE.equals(enableInfluxdb)) {
-            // nothing to do
             RepositoryService repositoryService = RepositoryStrategyFactory.get(StrategyConstant.Storage.INFLUXDB);
             savePointValueToRepository(pointValue, repositoryService);
         }
 
+        // 保存单个数据到 TDengine
+        if (Boolean.TRUE.equals(enableTDengine)) {
+            RepositoryService repositoryService = RepositoryStrategyFactory.get(StrategyConstant.Storage.TDENGINE);
+            savePointValueToRepository(pointValue, repositoryService);
+        }
+
         // 保存单个数据到 Opentsdb
         if (Boolean.TRUE.equals(enableOpentsdb)) {
             RepositoryService repositoryService = RepositoryStrategyFactory.get(StrategyConstant.Storage.STRATEGY_OPENTSDB);
@@ -89,7 +96,12 @@ public void save(List<PointValue> pointValues) {
 
             // 保存批量数据到 Influxdb
             if (Boolean.TRUE.equals(enableInfluxdb)) {
-                // nothing to do
+                RepositoryService repositoryService = RepositoryStrategyFactory.get(StrategyConstant.Storage.INFLUXDB);
+                savePointValuesToRepository(deviceId, values, repositoryService);
+            }
+
+            // 保存批量数据到 tdengine
+            if (Boolean.TRUE.equals(enableTDengine)) {
                 RepositoryService repositoryService = RepositoryStrategyFactory.get(StrategyConstant.Storage.INFLUXDB);
                 savePointValuesToRepository(deviceId, values, repositoryService);
             }

diff --git a/...c/main/java/io/github/pnoker/center/data/service/impl/repository/TDengineServiceImpl.java b/...c/main/java/io/github/pnoker/center/data/service/impl/repository/TDengineServiceImpl.java
@@ -0,0 +1,56 @@
+package io.github.pnoker.center.data.service.impl.repository;
+
+import io.github.pnoker.center.data.mapper.TaosPointValueMapper;
+import io.github.pnoker.center.data.service.RepositoryService;
+import io.github.pnoker.center.data.strategy.RepositoryStrategyFactory;
+import io.github.pnoker.common.constant.driver.StrategyConstant;
+import io.github.pnoker.common.entity.point.PointValue;
+import io.github.pnoker.common.entity.point.TaosPointValue;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+@ConditionalOnProperty(name = "data.point.sava.tdengine.enable", havingValue = "true")
+public class TDengineServiceImpl implements RepositoryService, InitializingBean {
+
+    @Resource
+    TaosPointValueMapper taosPointValueMapper;
+
+    @EventListener
+    public void initDatabase(ContextRefreshedEvent event) {
+        taosPointValueMapper.createSuperTable();
+    }
+
+    @Override
+    public String getRepositoryName() {
+        return StrategyConstant.Storage.TDENGINE;
+    }
+
+    @Override
+    public void savePointValue(PointValue pointValue) throws IOException {
+        taosPointValueMapper.createDeviceTable(pointValue.getDeviceId(), pointValue.getPointId());
+        taosPointValueMapper.insertOne(new TaosPointValue(pointValue));
+    }
+
+    @Override
+    public void savePointValues(String deviceId, List<PointValue> pointValues) throws IOException {
+        taosPointValueMapper.createDeviceTable(deviceId, pointValues.get(0).getPointId());
+        taosPointValueMapper.batchInsert(pointValues.stream().map(TaosPointValue::new).collect(Collectors.toList()));
+
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        RepositoryStrategyFactory.put(StrategyConstant.Storage.TDENGINE, this);
+    }
+}
diff --git a/dc3-center/dc3-center-data/src/main/resources/application.yml b/dc3-center/dc3-center-data/src/main/resources/application.yml
@@ -46,6 +46,8 @@ data:
     sava:
       influxdb:
         enable: false
+      tdengine:
+        enable: false
       opentsdb:
         enable: false
         host: dc3-opentsdb