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 18:08:05 +08:00
parent 50d817091e
commit d9bd70860c
15 changed files with 213 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/handler/MessageHandler.java
+68 -66
View File
@@ -1,5 +1,6 @@
package com.ruoyi.rocketmq.handler;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.core.constant.SecurityConstants;
import com.ruoyi.common.core.domain.R;
import com.ruoyi.common.core.enums.MsgEnum;
@@ -896,69 +897,66 @@ public class MessageHandler {
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
String heartbeatCountKey = HEARTBEAT_COUNT_PREFIX + clientId;
String alertKey = HEARTBEAT_ALERT_PREFIX + clientId;
try {
// 记录处理前状态（调试用）
String prevStatus = redisTemplate.opsForValue().get(statusKey);
String prevTime = redisTemplate.opsForValue().get(timeKey);
String prevHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}",
clientId, prevStatus, prevTime, prevHeartbeatCount);
Boolean prevAlertStatus = redisTemplate.hasKey(alertKey);
log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
clientId, prevStatus, prevTime, prevHeartbeatCount, prevAlertStatus);
// 使用事务确保原子性操作
List<Object> transactionResults = redisTemplate.execute(new SessionCallback<List<Object>>() {
// 原子递增心跳计数（线程安全）
Long newHeartbeatCount = redisTemplate.opsForValue().increment(heartbeatCountKey);
// 使用事务更新状态和时间
redisTemplate.execute(new SessionCallback<Object>() {
@Override
public List<Object> execute(RedisOperations operations) throws DataAccessException {
public Object 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();
}
});
// 从事务结果中获取新的心跳次数
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);
// 检查是否之前有告警状态（心跳恢复检测）
if (Boolean.TRUE.equals(redisTemplate.hasKey(alertKey))) {
log.info("客户端ID: {} 检测到心跳恢复", clientId);
// 原子递增恢复计数
Long recoveryCount = redisTemplate.opsForValue().increment(recoveryCountKey);
log.debug("客户端ID: {} 恢复计数: {}", clientId, recoveryCount);
if (recoveryCount >= 2) {
// 达到2次恢复，清除告警状态
log.warn("客户端ID: {} 心跳恢复达到{}次，清除告警状态", clientId, recoveryCount);
insertHeartbeatLog(clientId, "2", "心跳恢复,设备在线状态改为在线");
// 清理告警相关key
redisTemplate.delete(alertKey);
redisTemplate.delete(recoveryCountKey);
// 修改资源状态为在线
updateResourceStatus(clientId, "1");
log.info("客户端ID: {} 告警状态已清除", clientId);
} else {
// 未达到2次，只记录恢复次数
log.info("客户端ID: {} 心跳恢复第{}次", clientId, recoveryCount);
}
}
log.debug("客户端ID: {} 心跳处理完成，重置状态和时间，当前心跳次数: {}", clientId, newHeartbeatCount);
// 只有达到3次心跳才执行数据库操作
if (newHeartbeatCount >= 3) {
log.debug("客户端ID: {} 达到3次心跳，开始执行数据库操作", clientId);
log.info("客户端ID: {} 达到{}次心跳，开始执行数据库操作", clientId, newHeartbeatCount);
// 添加逻辑节点标识
RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote();
@@ -985,8 +983,10 @@ public class MessageHandler {
rmAgentManagementService.updateRmAgentManagementByHardwareSn(updateResultData);
}
}
// 交换机在线状态改为在线
// 交换机在线状态改为在线（只在第3次心跳时执行）
if(newHeartbeatCount == 3){
log.info("客户端ID: {} 第3次心跳，更新交换机状态为在线", clientId);
RmSwitchManagementRemote rmSwitchManagementRemote = new RmSwitchManagementRemote();
rmSwitchManagementRemote.setClientId(clientId);
R<List<RmSwitchManagementRemote>> rmSwitchManagementRemoteListR = remoteRevenueConfigService.getSwitchNameByClientId(rmSwitchManagementRemote, SecurityConstants.INNER);
@@ -997,35 +997,23 @@ public class MessageHandler {
switchUpdate.setClientId(clientId);
switchUpdate.setOnlineStatus("1");
remoteRevenueConfigService.updateSwitchMsgByClientId(switchUpdate, SecurityConstants.INNER);
log.info("客户端ID: {} 交换机状态更新完成", clientId);
} else {
log.debug("客户端ID: {} 未找到对应的交换机信息", clientId);
}
}
// 可选：达到3次后重置计数，或者继续累加
// redisTemplate.delete(heartbeatCountKey);
} else {
log.debug("客户端ID: {} 当前心跳次数: {}，未达到3次，跳过数据库操作", clientId, newHeartbeatCount);
}
// 检查是否之前有告警状态
if (Boolean.TRUE.equals(redisTemplate.hasKey(HEARTBEAT_ALERT_PREFIX + clientId))) {
// 获取当前恢复次数
String recoveryCountStr = redisTemplate.opsForValue().get(recoveryCountKey);
int recoveryCount = (recoveryCountStr == null) ? 1 : Integer.parseInt(recoveryCountStr) + 1;
// 记录处理后状态（调试用）
String currentStatus = redisTemplate.opsForValue().get(statusKey);
String currentTime = redisTemplate.opsForValue().get(timeKey);
String currentHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
Boolean currentAlertStatus = redisTemplate.hasKey(alertKey);
log.debug("客户端ID: {} 处理后状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
clientId, currentStatus, currentTime, currentHeartbeatCount, currentAlertStatus);
if (recoveryCount == 2) {
// 达到2次恢复，执行状态修改
log.warn("客户端ID: {} 心跳恢复达到2次，修改设备状态为在线", clientId);
insertHeartbeatLog(clientId, "2", "心跳恢复,设备在线状态改为在线");
redisTemplate.delete(HEARTBEAT_ALERT_PREFIX + clientId);
redisTemplate.delete(recoveryCountKey); // 清除恢复计数器
// 修改资源状态
updateResourceStatus(clientId, "1");
} else {
// 未达到2次，只记录恢复次数
log.info("客户端ID: {} 心跳恢复第{}次", clientId, recoveryCount);
redisTemplate.opsForValue().set(recoveryCountKey, String.valueOf(recoveryCount));
}
}
} catch (Exception e) {
log.error("处理心跳消息异常, clientId: {}", clientId, e);
}
@@ -1243,11 +1231,25 @@ public class MessageHandler {
if(rsp.getResult() != null){
List<RspResultVo> resultVos = JsonDataParser.parseJsonData(rsp.getResult(), RspResultVo.class);
RspResultVo rspResultVo = resultVos.get(0);
JSONObject jsonObject = JSONObject.parseObject(rsp.getResult());
String resOut = jsonObject.getString("resOut");
// 判断失败返回结果是否包含脚本执行失败关键字
R<EpsBusinessScriptRemote> businessScriptMsgR = remoteRevenueConfigService.getBusinessScriptMsgByScriptId(Long.valueOf(rspResultVo.getScriptId()), SecurityConstants.AUTHORIZATION_HEADER);
// 构建脚本执行结果实体类
RmResourceRemote insertData = new RmResourceRemote();
insertData.setClientId(message.getClientId());
insertData.setScriptId(Long.valueOf(rspResultVo.getScriptId()));
insertData.setResultFlag(1);
if(businessScriptMsgR != null && businessScriptMsgR.getData() != null){
EpsBusinessScriptRemote businessScriptMsg = businessScriptMsgR.getData();
String failedKeywords = businessScriptMsg.getFailedKeywords();
if(resOut.contains(failedKeywords)){
insertData.setResultFlag(0);
}else {
insertData.setResultFlag(1);
}
}else{
insertData.setResultFlag(1);
}
insertData.setDescription(rsp.getResult());
insertData.setCreateTime(createTime);
// 执行插入sql