Merge branch 'main' of https://git.jsspisoft.com/ry-das

das/src/main/java/com/das/modules/cache/domain/DeviceInfoCache.java

@@ -34,6 +34,8 @@ public class DeviceInfoCache {
      */
     private Long iotModelId;
 
+    private String iotModelCode;
+
     /**
      * 制造商
      */

das/src/main/java/com/das/modules/cache/service/impl/EquipmentCacheImpl.java

@@ -1,11 +1,14 @@
 package com.das.modules.cache.service.impl;
 
+import cn.hutool.core.util.StrUtil;
 import com.das.modules.cache.domain.DeviceInfoCache;
 import com.das.modules.cache.service.EquipmentCache;
 import com.das.modules.equipment.entity.SysEquipment;
 import com.das.modules.equipment.mapper.SysEquipmentMapper;
+import com.das.modules.equipment.mapper.SysIotModelMapper;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -14,11 +17,13 @@ import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 
+@Slf4j
 @Service
 public class EquipmentCacheImpl implements EquipmentCache {
     @Autowired
     SysEquipmentMapper sysEquipmentMapper;
-
+    @Autowired
+    SysIotModelMapper sysIotModelMapper;
 
     /**
      * 设备CODE索引，用于通过设备CODE访问设备缓存信息
@@ -43,6 +48,12 @@ public class EquipmentCacheImpl implements EquipmentCache {
             deviceInfoCache.setMadeinFactory(equipment.getMadeinFactory());
             deviceInfoCache.setParentDeviceId(equipment.getParentEquipmentId());
             deviceInfoCache.setIotModelId(equipment.getIotModelId());
+            String iotModelCode = sysIotModelMapper.getIodModelCode(equipment.getIotModelId());
+            log.debug("iotModelId: {}", equipment.getIotModelId());
+            if (StrUtil.isNotBlank(iotModelCode)){
+                log.debug("iotModelCode: {}", iotModelCode);
+                deviceInfoCache.setIotModelCode(iotModelCode);
+            }
             //创建Code索引
             deviceCodeIndex.put(deviceInfoCache.getDeviceCode(),deviceInfoCache);
             //创建Id索引

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

@@ -277,6 +277,9 @@ public class TDEngineService {
 
     @Async
     public void updateYCHighValues(List<RTData> values, String iotModelCode) {
+        if (values.isEmpty()) {
+            return;
+        }
         StringBuilder sb = new StringBuilder(1024 * 1024);
         try (Connection conn = hikariDataSource.getConnection();
              Statement pstmt = conn.createStatement()) {
@@ -315,6 +318,9 @@ public class TDEngineService {
 
     @Async
     public void updateYCLowValues(List<RTData> values, String iotModelCode) {
+        if (values.isEmpty()) {
+            return;
+        }
         StringBuilder sb = new StringBuilder(1024 * 1024);
         try (Connection conn = hikariDataSource.getConnection();
              Statement pstmt = conn.createStatement()) {
@@ -353,6 +359,9 @@ public class TDEngineService {
 
     @Async
     public void updateCalFieldValues(List<CalculateRTData> values) {
+        if (values.isEmpty()) {
+            return;
+        }
         StringBuilder sb = new StringBuilder(1024 * 1024);
         try (Connection conn = hikariDataSource.getConnection();
              Statement pstmt = conn.createStatement()) {
@@ -387,6 +396,9 @@ public class TDEngineService {
 
     @Async
     public void updateDeviceEventValues(List<DeviceEventInfo> values) {
+        if (values.isEmpty()) {
+            return;
+        }
         StringBuilder sb = new StringBuilder(1024 * 1024);
         try (Connection conn = hikariDataSource.getConnection();
              Statement pstmt = conn.createStatement()) {

das/src/main/java/com/das/modules/equipment/mapper/SysIotModelMapper.java

@@ -31,6 +31,8 @@ public interface SysIotModelMapper extends BaseMapper<SysIotModel> {
 
     String getIotModel(Long id);
 
+    String getIodModelCode(Long id);
+
     List<SysIotModelField> getAllIotModelField(Long id);
 
     SysIotModelVo selectIotModelByCode(String code);

das/src/main/java/com/das/modules/node/command/AnalogDataCommand.java

@@ -1,26 +1,110 @@
 package com.das.modules.node.command;
 
+import com.das.common.exceptions.ServiceException;
+import com.das.modules.cache.domain.DeviceInfoCache;
+import com.das.modules.cache.service.CacheService;
+import com.das.modules.cache.service.IotModelCache;
+import com.das.modules.data.service.TDEngineService;
 import com.das.modules.node.constant.NodeConstant;
+import com.das.modules.node.domain.bo.RTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
 import com.das.modules.node.service.NodeMessageService;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.*;
+
 @Service(value = NodeConstant.ANALOG_DATA)
 @Slf4j
 public class AnalogDataCommand implements BaseCommand {
 
     @Autowired
-    NodeMessageService nodeMessageService;
+    CacheService cacheService;
+
+    @Autowired
+    TDEngineService tdEngineService;
+
+    @Autowired
+    RedisTemplate<String, Object> redisTemplate;
+
+    /**
+     * 执行命令方法
+     * 当接收到终端消息时，此方法被调用以处理数据
+     * 主要负责处理模拟量数据的接收和处理
+     *
+     * @param data 包含模拟量数据的TerminalMessage对象
+     */
     @Override
     public void doCommand(TerminalMessage data) {
+        if (log.isDebugEnabled()){
+            log.debug("收到实时数据[模拟量]");
+            log.debug("数据内容: {}", data.toJsonString());
+        }
         try {
-            //analogData值只存入redis
+            processAnalogData(data);
-            nodeMessageService.handleData(data);
 
         } catch (Exception e) {
-            log.error("解析数据异常", e);
+            log.error("处理实时数据时产生异常", e);
         }
     }
+
+    /**
+     * 处理实时数据
+     * @param data
+     */
+
+    private void processAnalogData(TerminalMessage data){
+        JsonNode dataNode = data.getData();
+        String deviceId = Optional.of(dataNode.get("deviceId")).orElseThrow(() -> new ServiceException("deviceId字段缺失")).asText();
+        Long dataTime = Optional.of(dataNode.get("dataTime")).orElseThrow( () -> new ServiceException("dataTime字段缺失")).asLong();
+        JsonNode values = dataNode.get("values");
+        JsonNode archiveValues = dataNode.get("archiveValues");
+        //排除空数据
+        if (values == null && archiveValues == null){
+            return;
+        }
+        //如果values有值，则存入redis
+        Map<String,Object> redisValues = new HashMap<>();
+        if (values != null && values.isObject()){
+            for (Iterator<String> it = values.fieldNames(); it.hasNext(); ) {
+                String valueName = it.next();
+                String key = String.format("RT:%s:%s", deviceId, valueName.toLowerCase());
+                redisValues.put(key, values.get(valueName));
+            }
+        }
+        DeviceInfoCache dev = cacheService.getEquipmentCache().getDeviceInfoCacheById(Long.parseLong(deviceId));
+        IotModelCache iotCache = cacheService.getIotModelCache();
+        //如果archiveValues有值，则存入td，同时也更新redis
+        if (archiveValues != null && archiveValues.isObject()){
+            RTData hiSpeedData = new RTData();
+            hiSpeedData.setDeviceId(Long.parseLong(deviceId));
+            hiSpeedData.setDataTime(dataTime);
+            Map<String,Object> hiSpeedValues = new HashMap<>();
+            RTData lowSpeedData = new RTData();
+            lowSpeedData.setDeviceId(Long.parseLong(deviceId));
+            lowSpeedData.setDataTime(dataTime);
+            Map<String,Object> lowSpeedValues = new HashMap<>();
+            for (Iterator<String> it = archiveValues.fieldNames(); it.hasNext(); ) {
+                //加入redis更新列表
+                String valueName = it.next();
+                String key = String.format("RT:%s:%s", deviceId, valueName.toLowerCase());
+                redisValues.put(key, archiveValues.get(valueName));
+                //加入td更新列表
+                if (iotCache.isHighSpeed(dev.getIotModelId(), valueName)) {
+                    hiSpeedValues.put(valueName, archiveValues.get(valueName));
+                } else if (iotCache.isLowSpeed(dev.getIotModelId(), valueName)) {
+                    lowSpeedValues.put(valueName, archiveValues.get(valueName));
+                }
+            }
+            hiSpeedData.setValues(hiSpeedValues);
+            lowSpeedData.setValues(lowSpeedValues);
+            tdEngineService.updateYCHighValues(List.of(hiSpeedData), dev.getIotModelCode());
+            tdEngineService.updateYCLowValues(List.of(lowSpeedData), dev.getIotModelCode());
+        }
+        redisTemplate.opsForValue().multiSet(redisValues);
+    }
 }

das/src/main/java/com/das/modules/node/command/DeviceEventCommand.java

@@ -1,25 +1,128 @@
 package com.das.modules.node.command;
 
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.das.common.utils.AdminRedisTemplate;
+import com.das.common.utils.StringUtils;
+import com.das.modules.cache.domain.DeviceInfoCache;
+import com.das.modules.cache.service.CacheService;
+import com.das.modules.data.domain.DeviceEventInfo;
+import com.das.modules.data.service.TDEngineService;
+import com.das.modules.data.service.impl.DataServiceImpl;
 import com.das.modules.node.constant.NodeConstant;
 import com.das.modules.node.domain.bo.TerminalMessage;
+import com.das.modules.node.domain.vo.DeviceEventVo;
 import com.das.modules.node.service.NodeMessageService;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
 @Service(value = NodeConstant.DEVICE_EVENT)
 @Slf4j
 public class DeviceEventCommand implements BaseCommand{
     @Autowired
-    private NodeMessageService nodeMessageService;
+    AdminRedisTemplate adminRedisTemplate;
+
+    @Autowired
+    private DataServiceImpl dataService;
+
+    @Autowired
+    TDEngineService tdEngineService;
+
+
+    @Autowired
+    CacheService cacheService;
 
     @Override
     public void doCommand(TerminalMessage data) {
+        log.debug("收到实时[事件告警]");
         try {
-            nodeMessageService.handleDeviceEvent(data);
+            handleDeviceEvent(data);
 
         } catch (Exception e) {
             log.error("解析数据异常", e);
         }
     }
+
+    private void handleDeviceEvent(TerminalMessage data) {
+
+        JsonNode dataEvent = data.getData();
+        ObjectMapper objectMapper = new ObjectMapper();
+        List<DeviceEventInfo> valueList = new ArrayList<>();
+
+        // 使用 TypeReference 来指定转换目标类型
+        List<DeviceEventVo> list = objectMapper.convertValue(dataEvent, new TypeReference<List<DeviceEventVo>>() {
+        });
+        log.debug("消息data转化deviceVo,{}", list);
+        for (DeviceEventVo item : list) {
+            DeviceInfoCache deviceInfoCache = cacheService.getEquipmentCache().getDeviceInfoCacheById(Long.valueOf(item.getDeviceId()));
+            Integer firstTriggeredCode = adminRedisTemplate.get(String.format("RT:%s:%s", item.getDeviceId(), "FirstTriggeredCode".toLowerCase()));
+            DeviceEventInfo deviceEventInfo = new DeviceEventInfo();
+            deviceEventInfo.setEventTime(item.getEventTime());
+            deviceEventInfo.setEventId(IdWorker.getId());
+            deviceEventInfo.setAttributeCode(item.getAttrCode());
+            deviceEventInfo.setDeviceId(item.getDeviceId());
+            deviceEventInfo.setDeviceName(deviceInfoCache.getDeviceName());
+            deviceEventInfo.setDeviceCode(deviceInfoCache.getDeviceCode());
+            deviceEventInfo.setFirstTriggeredCode(firstTriggeredCode);
+            String eventType = getEventType(item.getEventType());
+            String model = dataService.deviceModelMap.get(item.getDeviceId());
+            if (StringUtils.isEmpty(model)) {
+                log.debug("未查询到物模型code，设备id：{}", item.getDeviceId());
+            }
+            String fieldName = dataService.fieldCodeNameMap.get(model).get(item.getAttrCode());
+            if (StringUtils.isEmpty(fieldName)) {
+                log.debug("未查询到物模型属性code，设备id：{}", item.getDeviceId());
+            }
+            deviceEventInfo.setEventType(item.getEventType());
+            deviceEventInfo.setConfirmed(0);
+            if (!StringUtils.isEmpty(eventType) && eventType.equals("遥信变位")) {
+                String stateDesc = dataService.stateDescMap.get(model).get(item.getAttrCode());
+                if (item.getAttrValue().equals(0)) {
+                    deviceEventInfo.setEventText(item.getAttrCode() + fieldName + " 复归");
+                    if (StringUtils.isNotEmpty(stateDesc)) {
+                        List<String> descList = Arrays.stream(stateDesc.split("\\|")).toList();
+                        deviceEventInfo.setEventText(item.getAttrCode() + fieldName + descList.get(0));
+                    }
+                    deviceEventInfo.setEventLevel(0);
+                } else {
+                    deviceEventInfo.setEventText(item.getAttrCode() + fieldName + " 动作");
+                    if (StringUtils.isNotEmpty(stateDesc)) {
+                        List<String> descList = Arrays.stream(stateDesc.split("\\|")).toList();
+                        deviceEventInfo.setEventText(item.getAttrCode() + fieldName + descList.get(1));
+                    }
+                    Integer level = dataService.eventLevelMap.get(model).get(item.getAttrCode());
+                    log.debug("level:{}", level);
+                    log.debug("fieldname{}", fieldName);
+                    deviceEventInfo.setEventLevel(level == null ? 0 : level);
+                }
+            } else {
+                deviceEventInfo.setEventText(item.getAttrCode() + fieldName + eventType + ",属性值为：" + item.getAttrValue() + ",越限值为：" + item.getLimitValue());
+                deviceEventInfo.setEventLevel(1);
+            }
+            valueList.add(deviceEventInfo);
+        }
+        try {
+            tdEngineService.updateDeviceEventValues(valueList);
+        } catch (Exception e) {
+            log.error("事件信息存入Td失败,失败原因", e);
+        }
+    }
+
+    private String getEventType(int eventType) {
+        return switch (eventType) {
+            case 0 -> "遥信变位";
+            case 1 -> "越上限";
+            case 2 -> "越下限";
+            case 3 -> "越上上限";
+            case 4 -> "越下下限";
+            default -> "";
+        };
+    }
 }

das/src/main/java/com/das/modules/node/command/HeartbeatCommand.java

@@ -34,7 +34,7 @@ public class HeartbeatCommand implements BaseCommand{
      */
     @Override
     public void doCommand(TerminalMessage data) {
-        log.info("收到[heartbeat]报文");
+        log.debug("收到[heartbeat]报文");
 
         // 解析心跳报文中的数据信息
         JsonNode dataInfo = data.getData();

das/src/main/java/com/das/modules/node/command/HisHighSpeedCommand.java

@@ -1,26 +1,67 @@
 package com.das.modules.node.command;
 
+import com.das.modules.cache.domain.DeviceInfoCache;
+import com.das.modules.cache.service.CacheService;
+import com.das.modules.data.service.TDEngineService;
 import com.das.modules.node.constant.NodeConstant;
+import com.das.modules.node.domain.bo.RTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
 import com.das.modules.node.service.NodeMessageService;
+import com.fasterxml.jackson.databind.JsonNode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.*;
+
 @Service(value = NodeConstant.HIS_HIGH_SPEED_DATA)
 @Slf4j
 public class HisHighSpeedCommand implements BaseCommand {
 
     @Autowired
     NodeMessageService nodeMessageService;
+
+    @Autowired
+    TDEngineService tdEngineService;
+
+    @Autowired
+    CacheService cacheService;
+
     @Override
     public void doCommand(TerminalMessage data) {
+        log.debug("收到历史高频数据");
         try {
             //analogData值只存入redis
-            nodeMessageService.handleHighSpeed(data);
+            handleHighSpeed(data);
 
         } catch (Exception e) {
             log.error("解析数据异常", e);
         }
     }
+
+    private void handleHighSpeed(TerminalMessage data) {
+        JsonNode jsonNode = data.getData();
+        String deviceId = jsonNode.get("deviceId").asText();
+        JsonNode values = jsonNode.get("values");
+        Map<String, Object> keyValueMap = new HashMap<>();
+        // 根据设备ID获取对应的物模型code
+        DeviceInfoCache dev = cacheService.getEquipmentCache().getDeviceInfoCacheById(jsonNode.get("deviceId").asLong());
+
+        //High数据
+        Iterator<String> keysHigh = values.fieldNames();
+        while (keysHigh.hasNext()) {
+            String fieldName = keysHigh.next();
+            keyValueMap.put(fieldName, values.get(fieldName));
+
+        }
+        Long dataTime = jsonNode.get("dataTime").asLong();
+        List<RTData> highList = new ArrayList<>();
+        RTData rtHighData = RTData.builder()
+                .dataTime(dataTime)
+                .deviceId(Long.valueOf(deviceId))
+                .values(keyValueMap)
+                .build();
+        highList.add(rtHighData);
+        tdEngineService.updateYCHighValues(highList, dev.getIotModelCode());
+    }
 }

das/src/main/java/com/das/modules/node/command/HisLowSpeedCommand.java

@@ -1,26 +1,65 @@
 package com.das.modules.node.command;
 
+import com.das.modules.cache.domain.DeviceInfoCache;
+import com.das.modules.cache.service.CacheService;
+import com.das.modules.data.service.TDEngineService;
+import com.das.modules.equipment.mapper.SysIotModelMapper;
 import com.das.modules.node.constant.NodeConstant;
+import com.das.modules.node.domain.bo.RTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
-import com.das.modules.node.service.NodeMessageService;
+import com.fasterxml.jackson.databind.JsonNode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.*;
+
 @Service(value = NodeConstant.HIS_LOW_SPEED_DATA)
 @Slf4j
 public class HisLowSpeedCommand implements BaseCommand {
 
     @Autowired
-    NodeMessageService nodeMessageService;
+    TDEngineService tdEngineService;
+
+
+    @Autowired
+    CacheService cacheService;
     @Override
     public void doCommand(TerminalMessage data) {
+        log.debug("收到[历史低频数据]");
         try {
             //analogData值只存入redis
-            nodeMessageService.handleLowSpeed(data);
+           handleLowSpeed(data);
 
         } catch (Exception e) {
             log.error("解析数据异常", e);
         }
     }
+
+    private void handleLowSpeed(TerminalMessage data) {
+        JsonNode jsonNode = data.getData();
+        String deviceId = jsonNode.get("deviceId").asText();
+        JsonNode values = jsonNode.get("values");
+        Map<String, Object> keyValueMap = new HashMap<>();
+        // 根据设备ID获取对应的物模型code
+        DeviceInfoCache dev = cacheService.getEquipmentCache().getDeviceInfoCacheById(jsonNode.get("deviceId").asLong());
+
+        //Low数据
+        Iterator<String> keysLow = values.fieldNames();
+        while (keysLow.hasNext()) {
+            String fieldName = keysLow.next();
+            keyValueMap.put(fieldName, values.get(fieldName));
+
+        }
+        Long dataTime = jsonNode.get("dataTime").asLong();
+        List<RTData> highList = new ArrayList<>();
+        RTData rtHighData = RTData.builder()
+                .dataTime(dataTime)
+                .deviceId(Long.valueOf(deviceId))
+                .values(keyValueMap)
+                .build();
+        highList.add(rtHighData);
+        tdEngineService.updateYCLowValues(highList, dev.getIotModelCode());
+    }
+
 }

das/src/main/java/com/das/modules/node/command/StateDataCommand.java

@@ -1,26 +1,95 @@
 package com.das.modules.node.command;
 
+import com.das.common.exceptions.ServiceException;
+import com.das.modules.cache.domain.DeviceInfoCache;
+import com.das.modules.cache.service.CacheService;
+import com.das.modules.cache.service.IotModelCache;
+import com.das.modules.data.service.TDEngineService;
 import com.das.modules.node.constant.NodeConstant;
+import com.das.modules.node.domain.bo.RTData;
 import com.das.modules.node.domain.bo.TerminalMessage;
 import com.das.modules.node.service.NodeMessageService;
+import com.fasterxml.jackson.databind.JsonNode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.*;
+
 @Service(value = NodeConstant.STATE_DATA)
 @Slf4j
 public class StateDataCommand implements BaseCommand {
 
     @Autowired
-    NodeMessageService nodeMessageService;
+    CacheService cacheService;
+
+    @Autowired
+    RedisTemplate<String,Object> redisTemplate;
+
+    @Autowired
+    TDEngineService tdEngineService;
 
     @Override
     public void doCommand(TerminalMessage data) {
+        log.debug("收到实时数据[状态量]");
         try {
             //只存入redis
-            nodeMessageService.handleData(data);
+            processStateData(data);
         } catch (Exception e) {
             log.error("解析数据异常", e);
         }
     }
+
+    private void processStateData(TerminalMessage data) {
+        JsonNode dataNode = data.getData();
+        String deviceId = Optional.of(dataNode.get("deviceId")).orElseThrow(() -> new ServiceException("deviceId字段缺失")).asText();
+        Long dataTime = Optional.of(dataNode.get("dataTime")).orElseThrow( () -> new ServiceException("dataTime字段缺失")).asLong();
+        JsonNode values = dataNode.get("values");
+        JsonNode archiveValues = dataNode.get("archiveValues");
+        //排除空数据
+        if (values == null && archiveValues == null){
+            return;
+        }
+        //如果values有值，则存入redis
+        Map<String,Object> redisValues = new HashMap<>();
+        if (values != null && values.isObject()){
+            for (Iterator<String> it = values.fieldNames(); it.hasNext(); ) {
+                String valueName = it.next();
+                String key = String.format("RT:%s:%s", deviceId, valueName.toLowerCase());
+                redisValues.put(key, values.get(valueName));
+            }
+        }
+        DeviceInfoCache dev = cacheService.getEquipmentCache().getDeviceInfoCacheById(Long.parseLong(deviceId));
+        IotModelCache iotCache = cacheService.getIotModelCache();
+        //如果archiveValues有值，则存入td，同时也更新redis
+        if (archiveValues != null && archiveValues.isObject()){
+            RTData hiSpeedData = new RTData();
+            hiSpeedData.setDeviceId(Long.parseLong(deviceId));
+            hiSpeedData.setDataTime(dataTime);
+            Map<String,Object> hiSpeedValues = new HashMap<>();
+            RTData lowSpeedData = new RTData();
+            lowSpeedData.setDeviceId(Long.parseLong(deviceId));
+            lowSpeedData.setDataTime(dataTime);
+            Map<String,Object> lowSpeedValues = new HashMap<>();
+            for (Iterator<String> it = archiveValues.fieldNames(); it.hasNext(); ) {
+                //加入redis更新列表
+                String valueName = it.next();
+                String key = String.format("RT:%s:%s", deviceId, valueName.toLowerCase());
+                redisValues.put(key, archiveValues.get(valueName));
+                //加入td更新列表
+                if (iotCache.isHighSpeed(dev.getIotModelId(), valueName)) {
+                    hiSpeedValues.put(valueName, archiveValues.get(valueName));
+                } else if (iotCache.isLowSpeed(dev.getIotModelId(), valueName)) {
+                    lowSpeedValues.put(valueName, archiveValues.get(valueName));
+                }
+            }
+            hiSpeedData.setValues(hiSpeedValues);
+            lowSpeedData.setValues(lowSpeedValues);
+
+            tdEngineService.updateYCHighValues(List.of(hiSpeedData), dev.getIotModelCode());
+            tdEngineService.updateYCLowValues(List.of(lowSpeedData),dev.getIotModelCode());
+        }
+        redisTemplate.opsForValue().multiSet(redisValues);
+    }
 }

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

@@ -14,14 +14,6 @@ public interface NodeMessageService {
 
     JsonNode sendTerminalConfig(Long nodeId);
 
-    void handleData(TerminalMessage data);
-
-    void handleHighSpeed(TerminalMessage data);
-
-    void handleLowSpeed(TerminalMessage data);
-
-    void handleDeviceEvent(TerminalMessage data);
-
     /**
      * 向指定采集节点发送指令(无返回值)
      * @param nodeId 节点ID

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

@@ -238,193 +238,6 @@ public class NodeMessageServiceImpl extends TextWebSocketHandler implements Node
 
     }
 
-    @Override
-    public void handleData(TerminalMessage data) {
-        JsonNode jsonNode = data.getData();
-        log.debug("收到消息：{}",data.getData());
-        String deviceId = jsonNode.get("deviceId").asText();
-        JsonNode values = jsonNode.get("values");
-        JsonNode archiveValues = jsonNode.get("archiveValues");
-        Long dataTime = jsonNode.get("dataTime").asLong();
-        Map<String, Object> keyValueMap = new HashMap<>();
-        String modelCode = dataService.deviceModelMap.get(deviceId);
-        Set<String> highKey = dataService.highIotFieldMap.get(modelCode).keySet();
-        Set<String> lowKey = dataService.lowIotFieldMap.get(modelCode).keySet();
-        Map<String, Object> highSpeedValueMap = new HashMap<>();
-        Map<String, Object> lowSpeedValueMap = new HashMap<>();
-
-        //数据入redis
-        if (values != null){
-            Iterator<String> keysHigh = values.fieldNames();
-            while (keysHigh.hasNext()) {
-                String fieldName = keysHigh.next();
-                String key = String.format("RT:%s:%s", deviceId, fieldName.toLowerCase());
-                keyValueMap.put(key, values.get(fieldName));
-            }
-        }
-
-        if (archiveValues != null){
-            Iterator<String> archiveKeys = archiveValues.fieldNames();
-            while (archiveKeys.hasNext()) {
-                String fieldName = archiveKeys.next();
-                String key = String.format("RT:%s:%s", deviceId, fieldName.toLowerCase());
-                keyValueMap.put(key, archiveValues.get(fieldName));
-                if (highKey.contains(fieldName)) {
-                    highSpeedValueMap.put(fieldName, archiveValues.get(fieldName));
-                }
-                if (lowKey.contains(fieldName)) {
-                    lowSpeedValueMap.put(fieldName, archiveValues.get(fieldName));
-                }
-            }
-        }
-        //更新td
-        if (!highSpeedValueMap.isEmpty()) {
-            List<RTData> highSpeedData = new ArrayList<>();
-            RTData rtHighData = RTData.builder().dataTime(dataTime).deviceId(Long.valueOf(deviceId)).values(highSpeedValueMap).build();
-            highSpeedData.add(rtHighData);
-            tdEngineService.updateYCHighValues(highSpeedData, modelCode);
-        }
-
-        if (!lowSpeedValueMap.isEmpty()) {
-            List<RTData> lowSpeedData = new ArrayList<>();
-            RTData rtLowData = RTData.builder().dataTime(dataTime).deviceId(Long.valueOf(deviceId)).values(lowSpeedValueMap).build();
-            lowSpeedData.add(rtLowData);
-            tdEngineService.updateYCLowValues(lowSpeedData, modelCode);
-        }
-        adminRedisTemplate.mSet(keyValueMap);
-    }
-
-    @Override
-    public void handleHighSpeed(TerminalMessage data) {
-        JsonNode jsonNode = data.getData();
-        String deviceId = jsonNode.get("deviceId").asText();
-        JsonNode values = jsonNode.get("values");
-        Map<String, Object> keyValueMap = new HashMap<>();
-        // 根据设备ID获取对应的物模型属性
-        String iotModelCode = sysIotModelMapper.getIotModel(jsonNode.get("deviceId").asLong());
-
-        //High数据
-        Iterator<String> keysHigh = values.fieldNames();
-        while (keysHigh.hasNext()) {
-            String fieldName = keysHigh.next();
-            keyValueMap.put(fieldName, values.get(fieldName));
-
-        }
-        Long dataTime = jsonNode.get("dataTime").asLong();
-        List<RTData> highList = new ArrayList<>();
-        RTData rtHighData = RTData.builder()
-                .dataTime(dataTime)
-                .deviceId(Long.valueOf(deviceId))
-                .values(keyValueMap)
-                .build();
-        highList.add(rtHighData);
-        tdEngineService.updateYCHighValues(highList, iotModelCode);
-    }
-
-    @Override
-    public void handleLowSpeed(TerminalMessage data) {
-        JsonNode jsonNode = data.getData();
-        String deviceId = jsonNode.get("deviceId").asText();
-        JsonNode values = jsonNode.get("values");
-        Map<String, Object> keyValueMap = new HashMap<>();
-        // 根据设备ID获取对应的物模型属性
-        String iotModelCode = sysIotModelMapper.getIotModel(jsonNode.get("deviceId").asLong());
-
-        //Low数据
-        Iterator<String> keysLow = values.fieldNames();
-        while (keysLow.hasNext()) {
-            String fieldName = keysLow.next();
-            keyValueMap.put(fieldName, values.get(fieldName));
-
-        }
-        Long dataTime = jsonNode.get("dataTime").asLong();
-        List<RTData> highList = new ArrayList<>();
-        RTData rtHighData = RTData.builder()
-                .dataTime(dataTime)
-                .deviceId(Long.valueOf(deviceId))
-                .values(keyValueMap)
-                .build();
-        highList.add(rtHighData);
-        tdEngineService.updateYCLowValues(highList, iotModelCode);
-    }
-
-    @Override
-    public void handleDeviceEvent(TerminalMessage data) {
-
-        JsonNode dataEvent = data.getData();
-        ObjectMapper objectMapper = new ObjectMapper();
-        List<DeviceEventInfo> valueList = new ArrayList<>();
-
-        // 使用 TypeReference 来指定转换目标类型
-        List<DeviceEventVo> list = objectMapper.convertValue(dataEvent, new TypeReference<List<DeviceEventVo>>() {
-        });
-        log.debug("消息data转化deviceVo,{}", list);
-        for (DeviceEventVo item : list) {
-            DeviceInfoCache deviceInfoCache = cacheService.getEquipmentCache().getDeviceInfoCacheById(Long.valueOf(item.getDeviceId()));
-            Integer firstTriggeredCode = adminRedisTemplate.get(String.format("RT:%s:%s", item.getDeviceId(), "FirstTriggeredCode".toLowerCase()));
-            DeviceEventInfo deviceEventInfo = new DeviceEventInfo();
-            deviceEventInfo.setEventTime(item.getEventTime());
-            deviceEventInfo.setEventId(IdWorker.getId());
-            deviceEventInfo.setAttributeCode(item.getAttrCode());
-            deviceEventInfo.setDeviceId(item.getDeviceId());
-            deviceEventInfo.setDeviceName(deviceInfoCache.getDeviceName());
-            deviceEventInfo.setDeviceCode(deviceInfoCache.getDeviceCode());
-            deviceEventInfo.setFirstTriggeredCode(firstTriggeredCode);
-            String eventType = getEventType(item.getEventType());
-            String model = dataService.deviceModelMap.get(item.getDeviceId());
-            if (StringUtils.isEmpty(model)) {
-                log.debug("未查询到物模型code，设备id：{}", item.getDeviceId());
-            }
-            String fieldName = dataService.fieldCodeNameMap.get(model).get(item.getAttrCode());
-            if (StringUtils.isEmpty(fieldName)) {
-                log.debug("未查询到物模型属性code，设备id：{}", item.getDeviceId());
-            }
-            deviceEventInfo.setEventType(item.getEventType());
-            deviceEventInfo.setConfirmed(0);
-            if (!StringUtils.isEmpty(eventType) && eventType.equals("遥信变位")) {
-                String stateDesc = dataService.stateDescMap.get(model).get(item.getAttrCode());
-                if (item.getAttrValue().equals(0)) {
-                    deviceEventInfo.setEventText(item.getAttrCode() + fieldName + " 复归");
-                    if (StringUtils.isNotEmpty(stateDesc)) {
-                        List<String> descList = Arrays.stream(stateDesc.split("\\|")).toList();
-                        deviceEventInfo.setEventText(item.getAttrCode() + fieldName + descList.get(0));
-                    }
-                    deviceEventInfo.setEventLevel(0);
-                } else {
-                    deviceEventInfo.setEventText(item.getAttrCode() + fieldName + " 动作");
-                    if (StringUtils.isNotEmpty(stateDesc)) {
-                        List<String> descList = Arrays.stream(stateDesc.split("\\|")).toList();
-                        deviceEventInfo.setEventText(item.getAttrCode() + fieldName + descList.get(1));
-                    }
-                    Integer level = dataService.eventLevelMap.get(model).get(item.getAttrCode());
-                    log.debug("level:{}", level);
-                    log.debug("fieldname{}", fieldName);
-                    deviceEventInfo.setEventLevel(level == null ? 0 : level);
-                }
-            } else {
-                deviceEventInfo.setEventText(item.getAttrCode() + fieldName + eventType + ",属性值为：" + item.getAttrValue() + ",越限值为：" + item.getLimitValue());
-                deviceEventInfo.setEventLevel(1);
-            }
-            valueList.add(deviceEventInfo);
-        }
-        try {
-            tdEngineService.updateDeviceEventValues(valueList);
-        } catch (Exception e) {
-            log.error("事件信息存入Td失败,失败原因", e);
-        }
-    }
-
-    private String getEventType(int eventType) {
-        return switch (eventType) {
-            case 0 -> "遥信变位";
-            case 1 -> "越上限";
-            case 2 -> "越下限";
-            case 3 -> "越上上限";
-            case 4 -> "越下下限";
-            default -> "";
-        };
-    }
-
     @Override
     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
         String remoteIp = (String) session.getAttributes().getOrDefault(NodeConstant.REMOTE_IP, "");

das/src/main/resources/mapper/SysIotModelMapper.xml

@@ -77,6 +77,10 @@
         select sim.iot_model_code  from sys_iot_model sim left join sys_equipment se on sim.id  = se.iot_model_id
         where se.id = #{id}
     </select>
+
+    <select id="getIodModelCode" resultType="java.lang.String">
+            select sim.iot_model_code from sys_iot_model sim where sim.id = #{id}
+    </select>
     <select id="getAllIotModelField" resultType="com.das.modules.equipment.entity.SysIotModelField">
         select simf.attribute_name as attributeName, simf.attribute_code as attributeCode,simf.highspeed as highSpeed,simf.datatype as dataType,simf.attribute_type as attributeType,simf.level as level,simf.stateDesc as stateDesc from sys_iot_model_field simf where simf.iot_model_id = #{id} order by simf.attribute_code
     </select>