心跳监测内容修改，ip对应数据改为clientId对应硬件SN，监控模板，策略bug修改

ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/DeviceMessageHandler.java

@@ -1,26 +1,24 @@
 package com.ruoyi.rocketmq.handler;
 
-import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.ruoyi.common.core.constant.SecurityConstants;
 import com.ruoyi.common.core.domain.R;
 import com.ruoyi.common.core.enums.MsgEnum;
 import com.ruoyi.common.core.utils.DateUtils;
-import com.ruoyi.common.core.utils.StringUtils;
 import com.ruoyi.rocketmq.domain.*;
 import com.ruoyi.rocketmq.domain.vo.CollectDataVo;
 import com.ruoyi.rocketmq.domain.vo.RspVo;
 import com.ruoyi.rocketmq.service.*;
 import com.ruoyi.rocketmq.utils.JsonDataParser;
 import com.ruoyi.system.api.RemoteRevenueConfigService;
-import com.ruoyi.system.api.domain.AllInterfaceNameRemote;
 import com.ruoyi.system.api.domain.EpsInitialTrafficDataRemote;
 import com.ruoyi.system.api.domain.InitialSwitchInfoDetailsRemote;
 import com.ruoyi.system.api.domain.RmResourceRegistrationRemote;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -37,6 +35,7 @@ import java.util.stream.Collectors;
  */
 @Slf4j
 @Component
+@EnableScheduling
 public class DeviceMessageHandler {
 
     private final Map<String, Consumer<DeviceMessage>> messageHandlers = new HashMap<>();
@@ -47,7 +46,7 @@ public class DeviceMessageHandler {
     // 心跳告警
     private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:";
     String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:";
-    private static final long HEARTBEAT_TIMEOUT = 180000; // 3分钟超时
+    private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时
 
 
     @Autowired
@@ -507,75 +506,6 @@ public class DeviceMessageHandler {
             throw new RuntimeException("交换机data数据为空");
         }
     }
-    /**
-     * 交换机数据入库
-     * @param message
-     */
-    private void handleSwitchMessage(DeviceMessage message) {
-        List<InitialSwitchInfo> switchInfos = JsonDataParser.parseJsonData(message.getData(), InitialSwitchInfo.class);
-        if(!switchInfos.isEmpty()){
-            // 时间戳转换
-            long timestamp = switchInfos.get(0).getTimestamp();
-            long millis = timestamp * 1000;
-            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
-            String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime);
-            // 查询临时表信息，计算实际流量值
-            InitialSwitchInfoTemp temp = new InitialSwitchInfoTemp();
-            temp.setClientId(message.getClientId());
-            List<InitialSwitchInfoTemp> tempList = initialSwitchInfoTempService.selectInitialSwitchInfoTempList(temp);
-            if(!tempList.isEmpty()){
-                // 1. 构建快速查找的Map
-                Map<String, InitialSwitchInfoTemp> tempMap = tempList.stream()
-                        .collect(Collectors.toMap(
-                                InitialSwitchInfoTemp::getName,
-                                Function.identity(),
-                                (existing, replacement) -> existing
-                        ));
-
-                // 2. 预计算除数（避免重复创建对象）
-                BigDecimal divisor = new BigDecimal(300);
-
-                // 3. 计算速度
-                switchInfos.forEach(switchInfo -> {
-                    switchInfo.setClientId(message.getClientId());
-                    switchInfo.setCreateTime(createTime);
-                    InitialSwitchInfoTemp tempInfo = tempMap.get(switchInfo.getName());
-                    if (tempInfo != null) {
-                        // 计算inSpeed
-                        if (switchInfo.getInBytes() != null && tempInfo.getInBytes() != null) {
-                            BigDecimal inDiff = switchInfo.getInBytes().subtract(tempInfo.getInBytes());
-                            switchInfo.setInSpeed(inDiff.divide(divisor, 2, RoundingMode.HALF_UP));
-                        }
-
-                        // 计算outSpeed
-                        if (switchInfo.getOutBytes() != null && tempInfo.getOutBytes() != null) {
-                            BigDecimal outDiff = switchInfo.getOutBytes().subtract(tempInfo.getOutBytes());
-                            switchInfo.setOutSpeed(outDiff.divide(divisor, 2, RoundingMode.HALF_UP));
-                        }
-                    }
-                });
-            }else{
-                switchInfos.forEach(switchInfo -> {
-                    switchInfo.setClientId(message.getClientId());
-                    switchInfo.setCreateTime(createTime);
-                });
-            }
-            // 清空临时表对应switch信息
-            initialSwitchInfoTempService.truncateSwitchInfoTemp(message.getClientId());
-            // 临时表 用来计算inSpeed outSeppd
-            initialSwitchInfoTempService.batchInsertInitialSwitchInfoTemp(switchInfos);
-            // 初始交换机数据入库
-            initialSwitchInfoService.batchInsertInitialSwitchInfo(switchInfos);
-            // 业务表入库
-            InitialSwitchInfoDetailsRemote detailsRemote = new InitialSwitchInfoDetailsRemote();
-            detailsRemote.setClientId(message.getClientId());
-            detailsRemote.setStartTime(timeStr);
-            detailsRemote.setEndTime(timeStr);
-            remoteRevenueConfigService.autoSaveSwitchTraffic(detailsRemote, SecurityConstants.INNER);
-        }else{
-            throw new RuntimeException("交换机data数据为空");
-        }
-    }
 
     /**
      * 系统其他信息
@@ -609,27 +539,6 @@ public class DeviceMessageHandler {
             }
         }
     }
-    /**
-     * 系统数据入库
-     * @param message
-     */
-    private void handleSystemMessage(DeviceMessage message) {
-        List<InitialSystemInfo> systemInfos = JsonDataParser.parseJsonData(message.getData(), InitialSystemInfo.class);
-        if(!systemInfos.isEmpty()){
-            // 时间戳转换
-            long timestamp = systemInfos.get(0).getTimestamp();
-            long millis = timestamp * 1000;
-            Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
-            systemInfos.forEach(iface -> {
-                iface.setClientId(message.getClientId());
-                iface.setCreateTime(createTime);
-            });
-            // 初始系统数据入库
-            initialSystemInfoService.batchInsertInitialSystemInfo(systemInfos);
-        }else{
-            throw new RuntimeException("系统data数据为空");
-        }
-    }
     /**
      * 监听心跳
      * @param message
@@ -662,7 +571,7 @@ public class DeviceMessageHandler {
                         redisTemplate.delete(HEARTBEAT_ALERT_PREFIX + clientId);
                         redisTemplate.delete(recoveryCountKey); // 清除恢复计数器
                         // 修改资源状态
-                        getResourceMsg(clientId, "1");
+                        updateResourceStatus(clientId, "1");
                     } else {
                         // 未达到2次，只记录恢复次数
                         log.info("客户端ID: {} 心跳恢复第{}次", clientId, recoveryCount);
@@ -676,7 +585,7 @@ public class DeviceMessageHandler {
     }
 
     // 添加一个定时任务方法，定期检查心跳状态
-    @Scheduled(fixedRate = 60000) // 每分钟检查一次
+    @Scheduled(fixedRate = 30000) // 每30s检查一次
     public void checkHeartbeatStatus() {
         long currentTime = System.currentTimeMillis();
         // 获取所有客户端时间键
@@ -686,8 +595,14 @@ public class DeviceMessageHandler {
         for (String timeKey : timeKeys) {
             String clientId = timeKey.substring(HEARTBEAT_TIME_PREFIX.length());
             String statusKey = HEARTBEAT_STATUS_PREFIX + clientId;
+            String alertKey = HEARTBEAT_ALERT_PREFIX + clientId;
 
             try {
+                // 检查是否已经存在告警
+                String existingAlert = redisTemplate.opsForValue().get(alertKey);
+                if ("1".equals(existingAlert)) {
+                    continue; // 如果已有告警，跳过处理
+                }
                 String lastTimeStr = redisTemplate.opsForValue().get(timeKey);
                 if (lastTimeStr == null) continue;
 
@@ -701,12 +616,21 @@ public class DeviceMessageHandler {
 
                     log.warn("客户端ID: {} 心跳丢失，连续次数: {}", clientId, lostCount);
 
-                    if (lostCount == 3) {
+                    if (lostCount >= 3) {
                         insertHeartbeatLog(clientId, "3", "连续三次心跳丢失");
                         redisTemplate.opsForValue().set(HEARTBEAT_ALERT_PREFIX + clientId, "1");
+                        // 设置告警后删除timeKey和statusKey
+                        redisTemplate.delete(timeKey);
+                        redisTemplate.delete(statusKey);
+
+                        log.info("客户端ID: {} 已设置告警并清理心跳记录", clientId);
                         // 修改资源状态
-                        getResourceMsg(clientId, "0");
+                        updateResourceStatus(clientId, "0");
                     }
+                }else {
+                    // 如果心跳正常，重置丢失次数
+                    redisTemplate.opsForValue().set(statusKey, "0");
+                    log.debug("客户端ID: {} 心跳正常，重置丢失次数", clientId);
                 }
             } catch (Exception e) {
                 log.error("检查心跳状态异常, clientId: {}", clientId, e);
@@ -714,57 +638,15 @@ public class DeviceMessageHandler {
         }
     }
 
-    /**
-     * 修改资源在线状态
-     * @param clientId
-     * @param status
-     */
-    private void getResourceMsg(String clientId, String status){
-        String ipAddress = null;
-        AllInterfaceNameRemote interfaceNameRemote = new AllInterfaceNameRemote();
-        interfaceNameRemote.setClientId(clientId);
-
-        // 1. 先获取交换机IP
-        interfaceNameRemote.setResourceType("2");
-        R<AllInterfaceNameRemote> switchResult = remoteRevenueConfigService.getMsgByClientId(
-                interfaceNameRemote, SecurityConstants.INNER);
-
-        if (switchResult != null && switchResult.getData() != null &&
-                StringUtils.isNotEmpty(switchResult.getData().getSwitchIp())) {
-            // 更新交换机状态
-            ipAddress = switchResult.getData().getSwitchIp();
-            updateResourceStatus(ipAddress, status);
-
-            // 2. 再获取服务器IP
-            interfaceNameRemote.setResourceType("1");
-            R<AllInterfaceNameRemote> serverResult = remoteRevenueConfigService.getMsgByClientId(
-                    interfaceNameRemote, SecurityConstants.INNER);
-
-            if (serverResult != null && serverResult.getData() != null &&
-                    StringUtils.isNotEmpty(serverResult.getData().getServerIp())) {
-                // 更新服务器状态
-                updateResourceStatus(serverResult.getData().getServerIp(), status);
-            }
-        } else {
-            // 3. 如果没有交换机IP，只获取服务器IP
-            interfaceNameRemote.setResourceType("1");
-            R<AllInterfaceNameRemote> serverResult = remoteRevenueConfigService.getMsgByClientId(
-                    interfaceNameRemote, SecurityConstants.INNER);
-
-            if (serverResult != null && serverResult.getData() != null &&
-                    StringUtils.isNotEmpty(serverResult.getData().getServerIp())) {
-                // 更新服务器状态
-                updateResourceStatus(serverResult.getData().getServerIp(), status);
-            } else {
-                log.warn("未找到客户端ID: {} 对应的IP地址", clientId);
-            }
-        }
-    }
     // 更新资源状态的公共方法
-    private void updateResourceStatus(String ipAddress, String status) {
+    private void updateResourceStatus(String clientId, String status) {
+        log.info("开启更新资源状态========");
         RmResourceRegistrationRemote rmResourceRegistrationRemote = new RmResourceRegistrationRemote();
         rmResourceRegistrationRemote.setOnlineStatus(status);
-        rmResourceRegistrationRemote.setIpAddress(ipAddress);
+        if("0".equals(status)){
+            rmResourceRegistrationRemote.setRegistrationStatus(status);
+        }
+        rmResourceRegistrationRemote.setHardwareSn(clientId);
         remoteRevenueConfigService.updateStatusByResource(rmResourceRegistrationRemote, SecurityConstants.INNER);
     }
     // 插入心跳日志到数据库
@@ -835,24 +717,4 @@ public class DeviceMessageHandler {
             }
         }
     }
-    /**
-     * 注册应答处理
-     * @param message
-     */
-//    private void handleRegisterMessage(DeviceMessage message) {
-//        RspVo rspVo = handleResponseMessage(message);
-//        String clientId = message.getClientId();
-//        if (rspVo != null && rspVo.getResCode() == 1) {
-//            RmResourceRegistrationRemote rmResourceRegistrationRemote = new RmResourceRegistrationRemote();
-//            rmResourceRegistrationRemote.setRegistrationStatus("1");
-//            rmResourceRegistrationRemote.setHardwareSn(clientId);
-//            remoteRevenueConfigService.updateStatusByResource(rmResourceRegistrationRemote, SecurityConstants.INNER);
-//        }else{
-//            if(rspVo == null){
-//                log.error("注册失败:应答信息为null");
-//            }else{
-//                log.error("注册失败:{}",rspVo.getResMsg());
-//            }
-//        }
-//    }
 }