liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化告警日志、消息推送

Browse Source
This commit is contained in:
gaoyutao
2025-11-13 13:02:58 +08:00
parent 2910280746
commit 50d817091e
3 changed files with 135 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java
+124 -54
View File
@@ -53,6 +53,8 @@ public class MessageHandler {
// 心跳告警 // 心跳告警
private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:"; private static final String HEARTBEAT_ALERT_PREFIX = "heartbeat:alert:";
String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:"; String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:";
String HEARTBEAT_COUNT_PREFIX = "heartbeat:count:";
private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时 private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时
@@ -889,54 +891,120 @@ public class MessageHandler {
String clientId = message.getClientId(); String clientId = message.getClientId();
String version = heartbeat.getVersion(); String version = heartbeat.getVersion();
log.debug("处理心跳消息，客户端ID: {}, 时间: {}", clientId, heartbeat.getTimestamp()); 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<RmAgentManagement> 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存储状态 // 使用Redis存储状态
String statusKey = HEARTBEAT_STATUS_PREFIX + clientId; String statusKey = HEARTBEAT_STATUS_PREFIX + clientId;
String timeKey = HEARTBEAT_TIME_PREFIX + clientId; String timeKey = HEARTBEAT_TIME_PREFIX + clientId;
String recoveryCountKey = HEARTBEAT_RECOVERY_COUNT_PREFIX + clientId; String recoveryCountKey = HEARTBEAT_RECOVERY_COUNT_PREFIX + clientId;
String heartbeatCountKey = HEARTBEAT_COUNT_PREFIX + clientId; // 新增心跳次数统计key
try { try {
// 记录处理前状态（调试用） // 记录处理前状态（调试用）
String prevStatus = redisTemplate.opsForValue().get(statusKey); String prevStatus = redisTemplate.opsForValue().get(statusKey);
String prevTime = redisTemplate.opsForValue().get(timeKey); 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<Object>() { List<Object> transactionResults = redisTemplate.execute(new SessionCallback<List<Object>>() {
@Override @Override
public Object execute(RedisOperations operations) throws DataAccessException { public List<Object> execute(RedisOperations operations) throws DataAccessException {
operations.multi(); 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，设置最后心跳时间 // 重置丢失计数为0，设置最后心跳时间
operations.opsForValue().set(statusKey, "0"); operations.opsForValue().set(statusKey, "0");
operations.opsForValue().set(timeKey, String.valueOf(System.currentTimeMillis())); operations.opsForValue().set(timeKey, String.valueOf(System.currentTimeMillis()));
operations.opsForValue().set(heartbeatCountKey, String.valueOf(newHeartbeatCount));
return operations.exec(); 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<RmAgentManagement> 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<List<RmSwitchManagementRemote>> 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))) { if (Boolean.TRUE.equals(redisTemplate.hasKey(HEARTBEAT_ALERT_PREFIX + clientId))) {
@@ -1108,32 +1176,34 @@ public class MessageHandler {
rmAlarmLog.setAlarmTime(DateUtils.getNowDate()); rmAlarmLog.setAlarmTime(DateUtils.getNowDate());
rmAlarmLog.setAlarmType("1"); rmAlarmLog.setAlarmType("1");
// 查询此告警是否存在 // 查询此告警是否存在
RmAlarmLog query = new RmAlarmLog(); // RmAlarmLog query = new RmAlarmLog();
query.setClientId(clientId); // query.setClientId(clientId);
query.setAlarmType(AlarmTypeEnum.服务器下线.getCode()); // query.setAlarmType(AlarmTypeEnum.服务器下线.getCode());
List<RmAlarmLog> alarmLog = rmAlarmLogService.selectRmAlarmLogList(query); // List<RmAlarmLog> alarmLog = rmAlarmLogService.selectRmAlarmLogList(query);
if(alarmLog != null && !alarmLog.isEmpty()){ // if(alarmLog != null && !alarmLog.isEmpty()){
rmAlarmLogService.updateRmAlarmLog(rmAlarmLog); // rmAlarmLogService.updateRmAlarmLog(rmAlarmLog);
}else{ // }
rmAlarmLogService.insertRmAlarmLog(rmAlarmLog); rmAlarmLogService.insertRmAlarmLog(rmAlarmLog);
// 拿到消息模板
RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig(); // 拿到消息模板
rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode()); RmAlarmPushConfig rmAlarmPushConfig = new RmAlarmPushConfig();
rmAlarmPushConfig.setPushAlarmTypes(AlarmTypeEnum.服务器下线.getCode()); rmAlarmPushConfig.setPushMethod(PushMethodEnum.企业微信.getCode());
List<RmAlarmPushConfig> alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig); rmAlarmPushConfig.setPushAlarmTypes(AlarmTypeEnum.服务器下线.getCode());
if(alarmConfigList != null && !alarmConfigList.isEmpty()){ List<RmAlarmPushConfig> alarmConfigList = rmAlarmPushConfigService.selectRmAlarmPushConfigList(rmAlarmPushConfig);
for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) { if(alarmConfigList != null && !alarmConfigList.isEmpty()){
String contentTemplate = alarmPushConfig.getMessageContent(); for (RmAlarmPushConfig alarmPushConfig : alarmConfigList) {
String webhookUrl = alarmPushConfig.getPushAddress(); String contentTemplate = alarmPushConfig.getMessageContent();
Map<String, Object> alarmMap = new HashMap<>(); String webhookUrl = alarmPushConfig.getPushAddress();
alarmMap.put("告警时间", rmAlarmLog.getAlarmTime()); String[] phones = alarmPushConfig.getContactPhones().split(",");
alarmMap.put("管理网-公网IP", rmAlarmLog.getMgmPublicIp()); String content = rmAlarmLog.getAlarmContent();
alarmMap.put("告警类型", AlarmTypeEnum.服务器下线.getMsg()); Map<String, Object> alarmMap = new HashMap<>();
alarmMap.put("告警设备", rmAlarmLog.getClientId()); alarmMap.put("告警时间", rmAlarmLog.getAlarmTime());
alarmMap.put("告警内容", rmAlarmLog.getAlarmContent()); alarmMap.put("管理网-公网IP", rmAlarmLog.getMgmPublicIp());
// 推送消息到企业微信 alarmMap.put("告警类型", AlarmTypeEnum.服务器下线.getMsg());
WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap); alarmMap.put("告警设备", rmAlarmLog.getClientId());
} alarmMap.put("告警内容", content);
// 推送消息到企业微信
WeChatWorkBot.sendTemplateMessage(webhookUrl, contentTemplate, alarmMap, content, phones, false);
} }
} }
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/WeChatWorkBot.java
+10 -10
View File
@@ -30,14 +30,14 @@ public class WeChatWorkBot {
* @param webhookUrl webhook地址 * @param webhookUrl webhook地址
* @param template 消息模板 * @param template 消息模板
* @param fieldValues 字段值的映射 * @param fieldValues 字段值的映射
* @param mentionedList 被@的用户列表（手机号） * @param mentionedMobiles 被@的用户列表（手机号）
* @param mentionedAll 是否@所有人 * @param mentionedAll 是否@所有人
* @return 是否发送成功 * @return 是否发送成功
*/ */
public static boolean sendTemplateMessage(String webhookUrl, String template, public static boolean sendTemplateMessage(String webhookUrl, String template,
Map<String, Object> fieldValues, Map<String, Object> fieldValues,
String[] mentionedList, boolean mentionedAll) { String[] mentionedMobiles, boolean mentionedAll) {
return sendTemplateMessage(webhookUrl, template, fieldValues, null, mentionedList, mentionedAll); return sendTemplateMessage(webhookUrl, template, fieldValues, null, mentionedMobiles, mentionedAll);
} }
/** /**
@@ -46,16 +46,16 @@ public class WeChatWorkBot {
* @param template 消息模板 * @param template 消息模板
* @param fieldValues 字段值的映射 * @param fieldValues 字段值的映射
* @param defaultValue 未找到字段时的默认值 * @param defaultValue 未找到字段时的默认值
* @param mentionedList 被@的用户列表（手机号） * @param mentionedMobiles 被@的用户列表（手机号）
* @param mentionedAll 是否@所有人 * @param mentionedAll 是否@所有人
* @return 是否发送成功 * @return 是否发送成功
*/ */
public static boolean sendTemplateMessage(String webhookUrl, String template, public static boolean sendTemplateMessage(String webhookUrl, String template,
Map<String, Object> fieldValues, Object defaultValue, Map<String, Object> fieldValues, Object defaultValue,
String[] mentionedList, boolean mentionedAll) { String[] mentionedMobiles, boolean mentionedAll) {
try { try {
String actualContent = processTemplate(template, fieldValues, defaultValue); String actualContent = processTemplate(template, fieldValues, defaultValue);
return sendTextMessage(webhookUrl, actualContent, mentionedList, mentionedAll); return sendTextMessage(webhookUrl, actualContent, mentionedMobiles, mentionedAll);
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
return false; return false;
@@ -213,7 +213,7 @@ public class WeChatWorkBot {
* 发送文本消息（支持@功能） * 发送文本消息（支持@功能）
*/ */
public static boolean sendTextMessage(String webhookUrl, String content, public static boolean sendTextMessage(String webhookUrl, String content,
String[] mentionedList, boolean mentionedAll) { String[] mentionedMobiles, boolean mentionedAll) {
try { try {
Map<String, Object> message = new HashMap<>(); Map<String, Object> message = new HashMap<>();
message.put("msgtype", "text"); message.put("msgtype", "text");
@@ -222,9 +222,9 @@ public class WeChatWorkBot {
textContent.put("content", content); textContent.put("content", content);
if (mentionedAll) { if (mentionedAll) {
textContent.put("mentioned_list", new String[]{"@all"}); textContent.put("mentioned_mobile_list", new String[]{"@all"});
} else if (mentionedList != null && mentionedList.length > 0) { } else if (mentionedMobiles != null && mentionedMobiles.length > 0) {
textContent.put("mentioned_list", mentionedList); textContent.put("mentioned_mobile_list", mentionedMobiles);
} }
message.put("text", textContent); message.put("text", textContent);
ruoyi-rocketmq/src/main/resources/mapper/rocketmq/RmAlarmLogMapper.xml
+1
View File
@@ -30,6 +30,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<if test="alarmType != null and alarmType != ''"> and alarm_type = #{alarmType}</if> <if test="alarmType != null and alarmType != ''"> and alarm_type = #{alarmType}</if>
<if test="alarmContent != null and alarmContent != ''"> and alarm_content = #{alarmContent}</if> <if test="alarmContent != null and alarmContent != ''"> and alarm_content = #{alarmContent}</if>
</where> </where>
order by alarm_time desc
</select> </select>
<select id="selectRmAlarmLogById" parameterType="Long" resultMap="RmAlarmLogResult"> <select id="selectRmAlarmLogById" parameterType="Long" resultMap="RmAlarmLogResult">