计算量数据更新至td，redis

das/src/main/java/com/das/modules/node/domain/bo/CalculateRTData.java

@@ -0,0 +1,23 @@
+package com.das.modules.node.domain.bo;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class CalculateRTData {
+    @JsonSerialize(using = ToStringSerializer.class)
+    private Long deviceId;
+
+    private Long dataTime;
+
+    private Object dataValue;
+
+    private String iotModelField;
+}

das/src/main/java/com/das/modules/node/domain/bo/RTData.java

@@ -19,5 +19,6 @@ public class RTData {
     @JsonSerialize(using = ToStringSerializer.class)
     private Long deviceId;
     private Long dataTime;
+    //字段和值的key value
     private Map<String, Object> values;
 }

das/src/main/java/com/das/modules/node/service/DataService.java

@@ -1,8 +1,11 @@
 package com.das.modules.node.service;
 
+import com.das.modules.node.domain.bo.CalculateRTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
 import com.fasterxml.jackson.databind.JsonNode;
 
+import java.util.List;
+
 public interface DataService {
 
     void pushMessage(TerminalMessage msg);
@@ -15,4 +18,6 @@ public interface DataService {
     void handleHighSpeed(TerminalMessage data);
 
     void handleLowSpeed(TerminalMessage data);
+
+    void updateCalFieldData(List<CalculateRTData> values);
 }

das/src/main/java/com/das/modules/node/service/TDEngineService.java

@@ -4,7 +4,9 @@ import cn.hutool.core.collection.ListUtil;
 import cn.hutool.core.util.StrUtil;
 import com.das.modules.equipment.domain.vo.IotModelFieldVo;
 import com.das.modules.equipment.mapper.SysIotModelMapper;
+import com.das.modules.node.domain.bo.CalculateRTData;
 import com.das.modules.node.domain.bo.RTData;
+import com.das.modules.node.service.impl.DataServiceImpl;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import jakarta.annotation.PreDestroy;
@@ -30,6 +32,8 @@ public class TDEngineService {
 
     private HikariDataSource hikariDataSource;
 
+    private DataServiceImpl dataService;
+
     @Value("${tdengine.url}")
     private String url;
 
@@ -336,6 +340,39 @@ public class TDEngineService {
         }
     }
 
+    @Async
+    public void updateCalFieldValues(List<CalculateRTData> values) {
+        StringBuilder sb = new StringBuilder(1024 * 1024);
+        try (Connection conn = hikariDataSource.getConnection();
+             Statement pstmt = conn.createStatement()) {
+            ListUtil.page(values, batchSize, (list) -> {
+                sb.setLength(0);
+                sb.append("insert into ");
+                for (CalculateRTData dv : list) {
+                    sb.append("c");
+                    sb.append(dv.getDeviceId());
+                    sb.append(" using c_");
+                    sb.append(dataService.deviceModelMap.get(dv.getDeviceId().toString()));
+                    sb.append("_").append(dv.getIotModelField());
+                    sb.append(" tags (");
+                    sb.append(dv.getDeviceId());
+                    sb.append(") values (");
+                    sb.append(dv.getDataTime());
+                    sb.append(",");
+                    sb.append(dv.getDataValue());
+                    sb.append(")");
+                }
+                try {
+                    pstmt.executeUpdate(sb.toString());
+                } catch (SQLException ex) {
+                    log.error("save yc error", ex);
+                }
+            });
+        } catch (SQLException ex) {
+            log.error(ex.getMessage());
+        }
+    }
+
     @Async
     public void updateStataValues(List<RTData> values, String iotModelCode) {
         StringBuilder sb = new StringBuilder(1024 * 1024);

das/src/main/java/com/das/modules/node/service/impl/DataServiceImpl.java

@@ -1,5 +1,6 @@
 package com.das.modules.node.service.impl;
 
+import cn.hutool.core.collection.ListUtil;
 import cn.hutool.core.util.IdUtil;
 import com.das.common.constant.MeasType;
 import com.das.common.utils.AdminRedisTemplate;
@@ -8,6 +9,7 @@ import com.das.modules.equipment.entity.SysIotModelField;
 import com.das.modules.equipment.mapper.SysIotModelMapper;
 import com.das.modules.node.disruptor.MessageEventFactory;
 import com.das.modules.node.disruptor.TerminalMessageEventHandler;
+import com.das.modules.node.domain.bo.CalculateRTData;
 import com.das.modules.node.domain.bo.RTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
 import com.das.modules.node.domain.vo.*;
@@ -44,6 +46,8 @@ public class DataServiceImpl implements DataService {
 
     private RingBuffer<TerminalMessage> ringBuffer = null;
 
+    public static final int COMMIT_NUMBER = 1000;
+
     @Resource
     TerminalMessageEventHandler terminalMessageEventHandler;
 
@@ -65,12 +69,19 @@ public class DataServiceImpl implements DataService {
     @Autowired
     TDEngineService tdEngineService;
 
+    //key:deviceId value:modelCode
+    public ConcurrentHashMap<String, String> deviceModelMap = new ConcurrentHashMap<>(10000);
+
+    //key:modelId  value:modelCode
     public ConcurrentHashMap<String, String> iotModelMap = new ConcurrentHashMap<>(10000);
 
+    //key:modelCode  value:Filed:Code,dataType
     public ConcurrentHashMap<String, Map<String, Object>> highIotFieldMap = new ConcurrentHashMap<>(10000);
 
+    //key:modelCode  value:Filed:Code,dataType
     public ConcurrentHashMap<String, Map<String, Object>> lowIotFieldMap = new ConcurrentHashMap<>(10000);
 
+    //key:modelCode  value:Filed:Code,dataType
     public ConcurrentHashMap<String, Map<String, String>> calculateIotFieldMap = new ConcurrentHashMap<>(10000);
 
     @PostConstruct
@@ -309,4 +320,18 @@ public class DataServiceImpl implements DataService {
         tdEngineService.updateYCLowValues(highList, iotModelCode);
     }
 
+    @Override
+    public void updateCalFieldData(List<CalculateRTData> calValues) {
+        //更新数据至redis,TD
+        ListUtil.page(calValues, COMMIT_NUMBER, list -> {
+            Map<String, Object> keyValueMap = new HashMap<>();
+            for (CalculateRTData value : list) {
+                String key = String.format("RT:%s:%s", value.getDeviceId(), value.getIotModelField().toLowerCase());
+                keyValueMap.put(key, value.getDataValue());
+            }
+            adminRedisTemplate.mSet(keyValueMap);
+            tdEngineService.updateCalFieldValues(list);
+        });
+    }
+
 }