转发逻辑修改，先转发、再处理逻辑；企业微信报警信息修改
parent
f47cf01cb0
commit
c9652702db
|
|
@ -1,5 +1,6 @@
|
|||
package com.agri.framework.interceptor;
|
||||
|
||||
import com.agri.common.utils.wechat.WxUtil;
|
||||
import com.agri.framework.config.MqttConfig;
|
||||
import com.agri.framework.manager.MqttAutoOffManager;
|
||||
import com.agri.framework.manager.MqttSubscriptionManager;
|
||||
|
|
@ -101,10 +102,9 @@ public class DeviceStatusHandler {
|
|||
/**
|
||||
* 处理设备状态：转发给订阅的前端、处理回执、触发自动关
|
||||
*/
|
||||
public void handle(String topic, String payload) throws MqttException {
|
||||
if (!payload.trim().startsWith("{")) {
|
||||
return;
|
||||
}
|
||||
public void handle(String topic, String payload) {
|
||||
|
||||
if (!isJsonObjectLike(payload)) return;
|
||||
// 第一步：解析JSON，非有效JSON直接return
|
||||
JSONObject payloadObj;
|
||||
try {
|
||||
|
|
@ -118,16 +118,23 @@ public class DeviceStatusHandler {
|
|||
return;
|
||||
}
|
||||
|
||||
// log.info("【设备处理】JSON解析：{}",payloadObj);
|
||||
// log.info("【设备处理】JSON解析：{}",payloadObj);
|
||||
// 解析设备ID：主题格式为dtu/{deviceId}/up，分割后第2个元素是设备ID
|
||||
String deviceId = topic.split("/")[1];
|
||||
String deviceId = extractDeviceId(topic);
|
||||
if (deviceId == null) return;
|
||||
// 转发消息
|
||||
forwardPayload(deviceId, payload);
|
||||
|
||||
isStartAutoOffTask(payloadObj,deviceId,payload);
|
||||
|
||||
}
|
||||
|
||||
private void isStartAutoOffTask(JSONObject payloadObj,String deviceId,String payload) {
|
||||
// 第二步：判断是否为设备回执（{"suc":true/false,"prop":{"功能码":指令}}）
|
||||
String funcType = null;
|
||||
Integer funcValue = null;
|
||||
boolean isAck = false;
|
||||
// 新增：标记是否需要执行自动关任务（全局可用）
|
||||
|
||||
// 第二步：设备回执处理逻辑（完全移除Redis写入）
|
||||
if (payloadObj.containsKey("suc") && payloadObj.containsKey("prop")) {
|
||||
isAck = true;
|
||||
|
|
@ -195,44 +202,72 @@ public class DeviceStatusHandler {
|
|||
// log.debug("【设备状态包】写入Redis成功，deviceId={}", deviceId);
|
||||
}
|
||||
}
|
||||
// 非回执消息：正常转发给订阅前端
|
||||
// 查询Redis中订阅该设备的前端列表：sub:{deviceId}
|
||||
Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);
|
||||
}
|
||||
|
||||
if (subscribedClients != null && !subscribedClients.isEmpty()) {
|
||||
// 推送给每个订阅的前端
|
||||
// 方案B：不再依赖online:；改为校验subc:{clientId}是否仍包含deviceId（取消订阅失败/异常退出兜底）
|
||||
List<String> clients = new ArrayList<>(subscribedClients);
|
||||
// 判断subc是否还存在 一次性查全部 获取失效的clientId
|
||||
List<Boolean> stillSubs = mqttSubscriptionManager.pipeIsMemberSubc(clients, deviceId);
|
||||
private void forwardPayload(String deviceId,String payload) {
|
||||
try {
|
||||
// 非回执消息：正常转发给订阅前端
|
||||
// 查询Redis中订阅该设备的前端列表：sub:{deviceId}
|
||||
Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);
|
||||
|
||||
// 关系不存在：清理sub:{deviceId}残留，避免一直给前端发
|
||||
List<String> stale = null;
|
||||
if (subscribedClients != null && !subscribedClients.isEmpty()) {
|
||||
// 推送给每个订阅的前端
|
||||
// 方案B：不再依赖online:；改为校验subc:{clientId}是否仍包含deviceId（取消订阅失败/异常退出兜底）
|
||||
List<String> clients = new ArrayList<>(subscribedClients);
|
||||
// 判断subc是否还存在 一次性查全部 获取失效的clientId
|
||||
List<Boolean> stillSubs = mqttSubscriptionManager.pipeIsMemberSubc(clients, deviceId);
|
||||
|
||||
for (int i = 0; i < clients.size(); i++) {
|
||||
String clientId = clients.get(i);
|
||||
boolean stillSub = i < stillSubs.size() && Boolean.TRUE.equals(stillSubs.get(i));
|
||||
if (!stillSub) {
|
||||
if (stale == null) {
|
||||
stale = new ArrayList<>();
|
||||
// 关系不存在：清理sub:{deviceId}残留，避免一直给前端发
|
||||
List<String> stale = null;
|
||||
|
||||
for (int i = 0; i < clients.size(); i++) {
|
||||
String clientId = clients.get(i);
|
||||
boolean stillSub = i < stillSubs.size() && Boolean.TRUE.equals(stillSubs.get(i));
|
||||
if (!stillSub) {
|
||||
if (stale == null) {
|
||||
stale = new ArrayList<>();
|
||||
}
|
||||
// false不存在添加队列
|
||||
stale.add(clientId);
|
||||
continue;
|
||||
}
|
||||
// false不存在添加队列
|
||||
stale.add(clientId);
|
||||
continue;
|
||||
// 前端专属主题：frontend/{clientId}/dtu/{deviceId}/listener
|
||||
String frontendTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/listener";
|
||||
// 发布消息
|
||||
mqttMessageSender.publish(frontendTopic, payload);
|
||||
// log.info("【设备状态转发】设备{} → 前端{}，主题：{}", deviceId, clientId, frontendTopic);
|
||||
}
|
||||
// 前端专属主题：frontend/{clientId}/dtu/{deviceId}/listener
|
||||
String frontendTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/listener";
|
||||
// 发布消息
|
||||
mqttMessageSender.publish(frontendTopic, payload);
|
||||
// log.info("【设备状态转发】设备{} → 前端{}，主题：{}", deviceId, clientId, frontendTopic);
|
||||
// 删掉设备对应的客户端
|
||||
if (stale != null && !stale.isEmpty()) {
|
||||
mqttSubscriptionManager.pipeSRemSub(deviceId, stale);
|
||||
}
|
||||
} else {
|
||||
// 优化：替换System.out为log.info
|
||||
// log.info("【设备状态转发】设备{}无订阅前端，跳过转发", deviceId);
|
||||
}
|
||||
// 删掉设备对应的客户端
|
||||
if (stale != null && !stale.isEmpty()) {
|
||||
mqttSubscriptionManager.pipeSRemSub(deviceId, stale);
|
||||
}
|
||||
} else {
|
||||
// 优化：替换System.out为log.info
|
||||
// log.info("【设备状态转发】设备{}无订阅前端，跳过转发", deviceId);
|
||||
} catch (MqttException e) {
|
||||
WxUtil.pushText(
|
||||
"【消息转发失败】\n deviceId: "+deviceId+"\n payload: "+payload+"\n msg: "+e.getMessage()+"\n cause: "+e.getCause());
|
||||
log.error("【消息转发失败】deviceId={}, msg={}", deviceId, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
private String extractDeviceId(String topic) {
|
||||
int first = topic.indexOf('/');
|
||||
if (first < 0) return null;
|
||||
int second = topic.indexOf('/', first + 1);
|
||||
if (second < 0) return null;
|
||||
return topic.substring(first + 1, second);
|
||||
}
|
||||
|
||||
private boolean isJsonObjectLike(String s) {
|
||||
if (s == null) return false;
|
||||
int n = s.length();
|
||||
for (int i = 0; i < n; i++) {
|
||||
char c = s.charAt(i);
|
||||
if (!Character.isWhitespace(c)) return c == '{';
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -175,7 +175,7 @@ public class MqttAutoOffManager {
|
|||
try {
|
||||
runAutoOff(deviceId, funcType);
|
||||
} catch (Exception e) {
|
||||
WxUtil.pushText("【自动关任务】提交任务失败! deviceId: "+deviceId+", funcType: "+funcType+", msg: "+e.getMessage());
|
||||
WxUtil.pushText("【自动关任务】提交任务失败! \n deviceId: "+deviceId+"\n funcType: "+funcType+"\n msg: "+e.getMessage()+"\n cause: "+e.getCause());
|
||||
log.error("【自动关任务】执行失败，deviceId={}, funcType={}", deviceId, funcType, e);
|
||||
} finally {
|
||||
// 任务执行完成后移除映射
|
||||
|
|
@ -207,7 +207,7 @@ public class MqttAutoOffManager {
|
|||
try {
|
||||
latestObj = JSON.parseObject(latest);
|
||||
} catch (Exception e) {
|
||||
WxUtil.pushText("自动关任务执行报错-解析异常：deviceId:" + deviceId + ", funcType:" + funcType+"异常："+e.getMessage());
|
||||
WxUtil.pushText("自动关任务执行报错-解析异常：\n deviceId: " + deviceId + "\n funcType:" + funcType+"\n 异常："+e.getMessage()+"\n Cause: "+e.getCause());
|
||||
log.warn("【自动关任务】最新状态JSON解析失败，跳过：deviceId={}, funcType={}", deviceId, funcType);
|
||||
return;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -142,7 +142,7 @@ public class MqttClientManager implements SmartLifecycle {
|
|||
|
||||
log.error("【MQTT连接异常】连接断开，clientId：{}，原因：{}",
|
||||
safeClientId(), (cause == null ? "unknown" : cause.getMessage()), cause);
|
||||
WxUtil.pushText("【MQTT连接异常】连接断开，clientId："+safeClientId()+"，原因："+(cause == null ? "unknown" : cause.getMessage()));
|
||||
WxUtil.pushText("【MQTT连接异常】连接断开：\n clientId："+safeClientId()+"\n 原因："+(cause == null ? "unknown" : cause.getMessage())+"\n Cause: "+cause.getCause());
|
||||
// 【方案A】不再触发自写重连；Paho自动重连会接管重连过程
|
||||
// 这里只记录日志即可
|
||||
if (isRunning.get()) {
|
||||
|
|
@ -170,7 +170,7 @@ public class MqttClientManager implements SmartLifecycle {
|
|||
// mqttBizPool.getActiveCount(),
|
||||
// mqttBizPool.getQueue().size());
|
||||
if (mqttBizPool.getActiveCount()>10 || mqttBizPool.getQueue().size()>1000) {
|
||||
WxUtil.pushText("线程池繁忙 正在处理中任务："+mqttBizPool.getActiveCount()+", 剩余待进行任务："+mqttBizPool.getQueue().size());
|
||||
WxUtil.pushText("线程池繁忙! \n 正在处理中任务："+mqttBizPool.getActiveCount()+"\n 剩余待进行任务："+mqttBizPool.getQueue().size());
|
||||
}
|
||||
try {
|
||||
// 优化：显式指定UTF-8编码，避免乱码（JDK 8兼容）
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ public class MqttMessageDispatcher {
|
|||
else if (topic.matches("frontend/\\w+/online")) {
|
||||
frontendOnlineHandler.handle(topic, payload);
|
||||
}
|
||||
// todo 是否加回复主题？？
|
||||
} catch (Exception e) {
|
||||
log.error("【MQTT消息处理异常】topic={}", topic, e);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue