diff --git a/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java b/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java index efaac4b..060f02a 100644 --- a/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java +++ b/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java @@ -53,6 +53,8 @@ public class MessageHandler { // 心跳告警 private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:"; String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:"; + + String HEARTBEAT_COUNT_PREFIX = "heartbeat:count:"; private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时 @@ -889,54 +891,120 @@ public class MessageHandler { String clientId = message.getClientId(); String version = heartbeat.getVersion(); log.debug("处理心跳消息，客户端ID: {}, 时间: {}", clientId, heartbeat.getTimestamp()); - // 添加逻辑节点标识 - RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote(); - updateData.setClientId(message.getClientId()); - updateData.setLogicalNodeId(heartbeat.getLogicalNode()); - updateData.setOnlineStatus("1"); - updateData.setAgentVersion(version); - updateData.setOnboardTime(DateUtils.getNowDate()); - remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER); - // agent更新结果存储 - RmAgentManagement query = new RmAgentManagement(); - query.setClientId(clientId); - List agentManagements = rmAgentManagementService.selectRmAgentManagementList(query); - if(!agentManagements.isEmpty()){ - RmAgentManagement rmAgentManagement = agentManagements.get(0); - if(!StringUtils.equals(rmAgentManagement.getAgentVersion(), version)){ - // 存储更新结果 - RmAgentManagement updateResultData = new RmAgentManagement(); - updateResultData.setId(rmAgentManagement.getId()); - updateResultData.setAgentVersion(version); - updateResultData.setLastUpdateTime(DateUtils.getNowDate()); - updateResultData.setLastUpdateResult("1"); - rmAgentManagementService.updateRmAgentManagementByHardwareSn(updateResultData); - } - } + // 使用Redis存储状态 String statusKey = HEARTBEAT_STATUS_PREFIX + clientId; String timeKey = HEARTBEAT_TIME_PREFIX + clientId; String recoveryCountKey = HEARTBEAT_RECOVERY_COUNT_PREFIX + clientId; + String heartbeatCountKey = HEARTBEAT_COUNT_PREFIX + clientId; // 新增心跳次数统计key try { // 记录处理前状态（调试用） String prevStatus = redisTemplate.opsForValue().get(statusKey); String prevTime = redisTemplate.opsForValue().get(timeKey); - log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}", clientId, prevStatus, prevTime); + String prevHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey); + log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}", + clientId, prevStatus, prevTime, prevHeartbeatCount); // 使用事务确保原子性操作 - redisTemplate.execute(new SessionCallback() { + List transactionResults = redisTemplate.execute(new SessionCallback>() { @Override - public Object execute(RedisOperations operations) throws DataAccessException { + public List execute(RedisOperations operations) throws DataAccessException { operations.multi(); + + // 获取当前心跳次数 + Object currentCountObj = operations.opsForValue().get(heartbeatCountKey); + int currentCount = 0; + if (currentCountObj != null) { + try { + currentCount = Integer.parseInt(currentCountObj.toString()); + } catch (NumberFormatException e) { + currentCount = 0; + } + } + + // 心跳次数+1 + int newHeartbeatCount = currentCount + 1; + // 重置丢失计数为0，设置最后心跳时间 operations.opsForValue().set(statusKey, "0"); operations.opsForValue().set(timeKey, String.valueOf(System.currentTimeMillis())); + operations.opsForValue().set(heartbeatCountKey, String.valueOf(newHeartbeatCount)); + return operations.exec(); } }); - log.debug("客户端ID: {} 心跳处理完成，重置状态和时间", clientId); + // 从事务结果中获取新的心跳次数 + int newHeartbeatCount = 0; + if (transactionResults != null && !transactionResults.isEmpty()) { + // 获取set操作的结果，最后一个set的值就是新的心跳次数 + try { + Object lastResult = transactionResults.get(transactionResults.size() - 1); + if (lastResult instanceof String) { + newHeartbeatCount = Integer.parseInt((String) lastResult); + } + } catch (Exception e) { + log.warn("获取事务结果失败，使用备用方式获取心跳次数", e); + // 备用方案：直接从Redis获取 + String countStr = redisTemplate.opsForValue().get(heartbeatCountKey); + if (countStr != null) { + newHeartbeatCount = Integer.parseInt(countStr); + } + } + } + + log.debug("客户端ID: {} 心跳处理完成，重置状态和时间，当前心跳次数: {}", clientId, newHeartbeatCount); + + // 只有达到3次心跳才执行数据库操作 + if (newHeartbeatCount >= 3) { + log.debug("客户端ID: {} 达到3次心跳，开始执行数据库操作", clientId); + + // 添加逻辑节点标识 + RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote(); + updateData.setClientId(message.getClientId()); + updateData.setLogicalNodeId(heartbeat.getLogicalNode()); + updateData.setOnlineStatus("1"); + updateData.setAgentVersion(version); + updateData.setOnboardTime(DateUtils.getNowDate()); + remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER); + + // agent更新结果存储 + RmAgentManagement query = new RmAgentManagement(); + query.setClientId(clientId); + List agentManagements = rmAgentManagementService.selectRmAgentManagementList(query); + if(!agentManagements.isEmpty()){ + RmAgentManagement rmAgentManagement = agentManagements.get(0); + if(!StringUtils.equals(rmAgentManagement.getAgentVersion(), version)){ + // 存储更新结果 + RmAgentManagement updateResultData = new RmAgentManagement(); + updateResultData.setId(rmAgentManagement.getId()); + updateResultData.setAgentVersion(version); + updateResultData.setLastUpdateTime(DateUtils.getNowDate()); + updateResultData.setLastUpdateResult("1"); + rmAgentManagementService.updateRmAgentManagementByHardwareSn(updateResultData); + } + } + // 交换机在线状态改为在线 + if(newHeartbeatCount == 3){ + RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote(); + rmSwitchManagementRemote.setClientId(clientId); + R> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER); + if(rmSwitchManagementRemoteListR != null && + rmSwitchManagementRemoteListR.getData()!=null && + !rmSwitchManagementRemoteListR.getData().isEmpty()){ + RmSwitchManagementRemote switchUpdate = new RmSwitchManagementRemote(); + switchUpdate.setClientId(clientId); + switchUpdate.setOnlineStatus("1"); + remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER); + } + } + + // 可选：达到3次后重置计数，或者继续累加 + // redisTemplate.delete(heartbeatCountKey); + } else { + log.debug("客户端ID: {} 当前心跳次数: {}，未达到3次，跳过数据库操作", clientId, newHeartbeatCount); + } // 检查是否之前有告警状态 if (Boolean.TRUE.equals(redisTemplate.hasKey(HEARTBEAT_ALERT_PREFIX + clientId))) { @@ -1108,32 +1176,34 @@ public class MessageHandler { rmAlarmLog.setAlarmTime(DateUtils.getNowDate()); rmAlarmLog.setAlarmType("1"); // 查询此告警是否存在 - RmAlarmLog query = new RmAlarmLog(); - query.setClientId(clientId); - query.setAlarmType(AlarmTypeEnum.服务器下线.getCode()); - List alarmLog = rmAlarmLogService.selectRmAlarmLogList(query); - if(alarmLog != null && !alarmLog.isEmpty()){ - rmAlarmLogService.updateRmAlarmLog(rmAlarmLog); - }else{ - rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); - // 拿到消息模板 - RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig(); - rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode()); - rmAlarmPushConfig.setPushAlarmTypes(AlarmTypeEnum.服务器下线.getCode()); - List alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig); - if(alarmConfigList != null && !alarmConfigList.isEmpty()){ - for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) { - String contentTemplate = alarmPushConfig.getMessageContent(); - String webhookUrl = alarmPushConfig.getPushAddress(); - Map alarmMap = new HashMap<>(); - alarmMap.put("告警时间", rmAlarmLog.getAlarmTime()); - alarmMap.put("管理网-公网IP", rmAlarmLog.getMgmPublicIp()); - alarmMap.put("告警类型", AlarmTypeEnum.服务器下线.getMsg()); - alarmMap.put("告警设备", rmAlarmLog.getClientId()); - alarmMap.put("告警内容", rmAlarmLog.getAlarmContent()); - // 推送消息到企业微信 - WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap); - } +// RmAlarmLog query = new RmAlarmLog(); +// query.setClientId(clientId); +// query.setAlarmType(AlarmTypeEnum.服务器下线.getCode()); +// List alarmLog = rmAlarmLogService.selectRmAlarmLogList(query); +// if(alarmLog != null && !alarmLog.isEmpty()){ +// rmAlarmLogService.updateRmAlarmLog(rmAlarmLog); +// } + rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); + + // 拿到消息模板 + RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig(); + rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode()); + rmAlarmPushConfig.setPushAlarmTypes(AlarmTypeEnum.服务器下线.getCode()); + List alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig); + if(alarmConfigList != null && !alarmConfigList.isEmpty()){ + for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) { + String contentTemplate = alarmPushConfig.getMessageContent(); + String webhookUrl = alarmPushConfig.getPushAddress(); + String[] phones = alarmPushConfig.getContactPhones().split(","); + String content = rmAlarmLog.getAlarmContent(); + Map alarmMap = new HashMap<>(); + alarmMap.put("告警时间", rmAlarmLog.getAlarmTime()); + alarmMap.put("管理网-公网IP", rmAlarmLog.getMgmPublicIp()); + alarmMap.put("告警类型", AlarmTypeEnum.服务器下线.getMsg()); + alarmMap.put("告警设备", rmAlarmLog.getClientId()); + alarmMap.put("告警内容", content); + // 推送消息到企业微信 + WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap, content, phones, false); } } diff --git a/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/WeChatWorkBot.java b/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/WeChatWorkBot.java index dca7b38..d489dfe 100644 --- a/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/WeChatWorkBot.java +++ b/ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/WeChatWorkBot.java @@ -30,14 +30,14 @@ public class WeChatWorkBot { * @param webhookUrl webhook地址 * @param template 消息模板 * @param fieldValues 字段值的映射 - * @param mentionedList 被@的用户列表（手机号） + * @param mentionedMobiles 被@的用户列表（手机号） * @param mentionedAll 是否@所有人 * @return 是否发送成功 */ public static boolean sendTemplateMessage(String webhookUrl, String template, Map fieldValues, - String[] mentionedList, boolean mentionedAll) { - return sendTemplateMessage(webhookUrl, template, fieldValues, null, mentionedList, mentionedAll); + String[] mentionedMobiles, boolean mentionedAll) { + return sendTemplateMessage(webhookUrl, template, fieldValues, null, mentionedMobiles, mentionedAll); } /** @@ -46,16 +46,16 @@ public class WeChatWorkBot { * @param template 消息模板 * @param fieldValues 字段值的映射 * @param defaultValue 未找到字段时的默认值 - * @param mentionedList 被@的用户列表（手机号） + * @param mentionedMobiles 被@的用户列表（手机号） * @param mentionedAll 是否@所有人 * @return 是否发送成功 */ public static boolean sendTemplateMessage(String webhookUrl, String template, Map fieldValues, Object defaultValue, - String[] mentionedList, boolean mentionedAll) { + String[] mentionedMobiles, boolean mentionedAll) { try { String actualContent = processTemplate(template, fieldValues, defaultValue); - return sendTextMessage(webhookUrl, actualContent, mentionedList, mentionedAll); + return sendTextMessage(webhookUrl, actualContent, mentionedMobiles, mentionedAll); } catch (Exception e) { e.printStackTrace(); return false; @@ -213,7 +213,7 @@ public class WeChatWorkBot { * 发送文本消息（支持@功能） */ public static boolean sendTextMessage(String webhookUrl, String content, - String[] mentionedList, boolean mentionedAll) { + String[] mentionedMobiles, boolean mentionedAll) { try { Map message = new HashMap<>(); message.put("msgtype", "text"); @@ -222,9 +222,9 @@ public class WeChatWorkBot { textContent.put("content", content); if (mentionedAll) { - textContent.put("mentioned_list", new String[]{"@all"}); - } else if (mentionedList != null && mentionedList.length > 0) { - textContent.put("mentioned_list", mentionedList); + textContent.put("mentioned_mobile_list", new String[]{"@all"}); + } else if (mentionedMobiles != null && mentionedMobiles.length > 0) { + textContent.put("mentioned_mobile_list", mentionedMobiles); } message.put("text", textContent); diff --git a/ruoyi-rocketmq/src/main/resources/mapper/rocketmq/RmAlarmLogMapper.xml b/ruoyi-rocketmq/src/main/resources/mapper/rocketmq/RmAlarmLogMapper.xml index 373825e..c8fc58f 100644 --- a/ruoyi-rocketmq/src/main/resources/mapper/rocketmq/RmAlarmLogMapper.xml +++ b/ruoyi-rocketmq/src/main/resources/mapper/rocketmq/RmAlarmLogMapper.xml @@ -30,6 +30,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" and alarm_type = #{alarmType} and alarm_content = #{alarmContent} + order by alarm_time desc