root
/
agri
2
0
Fork 0
Code Issues Pull Requests Packages Projects 1 Releases Wiki Activity

mqtt基础功能

feasure
xce 2026-01-17 00:46:50 +08:00
parent bd24cec8cc
commit e7b2f43cf0
2 changed files with 125 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
agri-admin/src/main/resources/application-mqtt.yml
View File

@ -2,14 +2,11 @@ spring:
# MQTT配置
mqtt:
host: tcp://122.51.109.52:1883 # 设备/后端的MQTT TCP地址
ws-host: wss://mq.xiaoces.com/mqtt # 前端的WebSocket地址
username: admin # Mosquitto共用账号
password: Admin#12345678 # Mosquitto密码
client-id: springboot-backend # 截取UUID前8位（自动去横线）
default-topic: dtu/+/up,frontend/+/control/+ # 后端监听的主题
default-topic: dtu/+/up,frontend/+/down/+ # 后端监听的主题
qos: 1 # 消息可靠性
timeout: 60 # 连接超时
keep-alive: 60 # 心跳间隔
# 新增重连配置
reconnect-interval: 5 # 重连间隔（秒）
max-reconnect-times: -1 # 最大重连次数（-1=无限重连）

159
agri-framework/src/main/java/com/agri/framework/interceptor/MqttMessageHandler.java
View File

@ -1,6 +1,9 @@
package com.agri.framework.interceptor;
import com.agri.framework.config.MqttConfig;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
@ -25,6 +28,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -37,7 +41,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
* 3. 转发设备状态到订阅的前端
* 4. 处理前端控制指令（权限校验+分布式锁+转发）
* 适配JDK 8，无心跳包相关逻辑
*
* <p>
* 【方案A改造说明（最小改动）】
* 1) 不再自己new MqttClient 做重连，避免 mqttMessageSender 持有旧client导致“重连后发不出去”
* 2) 只依赖 MqttConnectOptions#setAutomaticReconnect(true) 的Paho自动重连
@ -46,20 +50,26 @@ import java.util.concurrent.atomic.AtomicBoolean;
@Component
public class MqttMessageHandler implements SmartLifecycle {
/** MQTT客户端（由MqttConfig配置类注入） */
/**
* MQTT客户端（由MqttConfig配置类注入）
*/
@Resource
private MqttClient mqttClient;
/** MQTT消息发送工具类（由MqttConfig配置类注入） */
/**
* MQTT消息发送工具类（由MqttConfig配置类注入）
*/
@Resource
private MqttConfig.MqttMessageSender mqttMessageSender;
/** Redis模板，用于存储订阅关系、设备在线状态、分布式锁 */
/**
* Redis模板，用于存储订阅关系、设备在线状态、分布式锁
*/
@Resource
private StringRedisTemplate stringRedisTemplate;
// 读取配置文件中的默认订阅主题（移除心跳主题）
@Value("${spring.mqtt.default-topic:dtu/+/up,frontend/+/control/+}")
@Value("${spring.mqtt.default-topic:dtu/+/up,frontend/+/down/+}")
private String defaultTopic;
// 优化：统一使用SLF4J日志（JDK 8兼容）
@ -68,14 +78,16 @@ public class MqttMessageHandler implements SmartLifecycle {
// 新增：生命周期管理标识，控制MQTT客户端启动/关闭
private final AtomicBoolean isRunning = new AtomicBoolean(false);
/** MQTT连接配置项（从MqttConfig注入） */
/**
* MQTT连接配置项（从MqttConfig注入）
*/
@Resource
private MqttConnectOptions mqttConnectOptions;
/**
* 初始化：订阅主题+设置回调
* （移除@PostConstruct，改为由SmartLifecycle的start()触发）
*
* <p>
* 【方案A】不做自写重连；Paho会在连接断开后自动重连（前提：connectOptions.setAutomaticReconnect(true)）
*/
public void subscribeTopics() throws MqttException {
@ -104,6 +116,7 @@ public class MqttMessageHandler implements SmartLifecycle {
// 按主题类型设置QoS：控制指令/状态用QoS 1
for (int i = 0; i < topics.length; i++) {
qosArray[i] = 1;
topics[i] = topics[i].trim();
}
// 设置MQTT消息回调：处理连接断开、消息接收、消息发布完成
@ -170,6 +183,7 @@ public class MqttMessageHandler implements SmartLifecycle {
/**
* 消息分发处理：根据主题类型路由到不同处理方法\仅处理设备状态、前端控制指令
* 可以监听到设备传过来的业务数据 以及前端传过来的控制设备指令
*
* @param topic 消息主题
* @param payload 消息内容（JSON字符串）
*/
@ -182,8 +196,8 @@ public class MqttMessageHandler implements SmartLifecycle {
if (topic.matches("dtu/\\w+/up")) {
handleDeviceStatus(topic, payload);
}
// 处理前端控制指令主题：frontend/{clientId}/control/{deviceId}
else if (topic.matches("frontend/\\w+/control/\\w+")) {
// 处理前端控制指令主题：frontend/{clientId}/down/{deviceId}
else if (topic.matches("frontend/\\w+/down/\\w+")) {
handleFrontendControl(topic, payload);
}
} catch (Exception e) {
@ -196,8 +210,51 @@ public class MqttMessageHandler implements SmartLifecycle {
* 处理设备状态：转发给订阅的前端
*/
private void handleDeviceStatus(String topic, String payload) throws MqttException {
// 第一步：解析JSON，非有效JSON直接return
JSONObject payloadObj;
try {
payloadObj = JSON.parseObject(payload);
} catch (Exception e) {
log.error("【设备处理】JSON解析失败，payload={}", payload, e);
return;
}
if (payloadObj == null || payloadObj.isEmpty()) {
log.warn("【设备处理】JSON解析后为空，payload={}", payload);
return;
}
// 解析设备ID：主题格式为dtu/{deviceId}/up，分割后第2个元素是设备ID
String deviceId = topic.split("/")[1];
// 第二步：判断是否为设备回执（{"suc":true/false,"prop":{"功能码":指令}}）
String funcType = null;
if (payloadObj.containsKey("suc") && payloadObj.containsKey("prop")) {
JSONObject propObj = payloadObj.getJSONObject("prop");
if (propObj != null && !propObj.isEmpty()) {
// 提取prop中的第一个功能码
Map.Entry<String, Object> propEntry = propObj.entrySet().iterator().next();
funcType = propEntry.getKey();
// 释放对应功能的分布式锁
String lockKey = "lock:" + deviceId + ":" + funcType;
Boolean delete = stringRedisTemplate.delete(lockKey);
if (propObj.size() > 1) {
log.warn("【设备回执】prop包含多个功能码，仅处理第一个：{}", propObj.keySet());
}
log.info("【设备回执】设备{}的{}功能执行完成，已释放锁：{},{}", deviceId, funcType, lockKey, delete);
// 广播回执结果给所有订阅该设备的前端
// String broadcastTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/up";;
// JSONObject ackPayload = new JSONObject();
// ackPayload.put("deviceId", deviceId);
// ackPayload.put("funcType", funcType);
// ackPayload.put("suc", payloadObj.getBooleanValue("suc"));
// ackPayload.put("code", propEntry.getValue());
// mqttMessageSender.publish(broadcastTopic, ackPayload.toJSONString());
}
}
// 非回执消息：正常转发给订阅前端
// if (!isDeviceAck) {
// 查询Redis中订阅该设备的前端列表：sub:{deviceId}
Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);
@ -215,6 +272,7 @@ public class MqttMessageHandler implements SmartLifecycle {
// 优化：替换System.out为log.info
log.info("【设备状态转发】设备{}无订阅前端，跳过转发", deviceId);
}
// }
}
/**
@ -232,37 +290,57 @@ public class MqttMessageHandler implements SmartLifecycle {
return;
}
// 解析功能码（{"功能码":状态码}格式）
Map<String, Integer> funcCodeMap = null;
try {
funcCodeMap = JSON.parseObject(payload, new TypeReference<Map<String, Integer>>() {
});
} catch (Exception e) {
log.error("【指令处理】功能码解析失败，payload={}", payload, e);
// String errorTopic = "frontend/" + clientId + "/dtu/" + deviceId+"/up";
// mqttMessageSender.publish(errorTopic, "{\"msg\":\"指令格式错误\"}");
return;
}
if (funcCodeMap == null || funcCodeMap.isEmpty()) {
// String errorTopic = "frontend/" + clientId + "/dtu/" + deviceId+"/up";
// mqttMessageSender.publish(errorTopic, "{\"msg\":\"功能码不能为空\"}");
log.warn("【指令处理】前端{}操作设备{}失败：功能码为空", clientId, deviceId);
return;
}
// 提取第一个功能码作为锁标识
String funcType = funcCodeMap.keySet().iterator().next();
// 1. 权限校验（示例：admin开头有全权限）
if (!checkPermission(clientId, deviceId)) {
String errorTopic = "frontend/" + clientId + "/error/" + deviceId;
String errorTopic = "frontend/" + clientId + "/dtu/" + deviceId+"/up";
mqttMessageSender.publish(errorTopic, "{\"msg\":\"无设备操作权限\"}");
// 优化：替换System.err为log.warn
log.warn("【权限校验】前端{}操作设备{}失败", clientId, deviceId);
return;
}
// 2. 分布式锁：避免多前端并发控制
String lockKey = "lock:" + deviceId;
// 2. 分布式锁：设备ID+功能类型（避免同设备同功能并发控制）
String lockKey = "lock:" + deviceId + ":" + funcType;
Boolean lockSuccess = stringRedisTemplate.opsForValue().setIfAbsent(
lockKey, clientId, 10, TimeUnit.SECONDS // 优化：显式指定时间单位
lockKey, clientId, 15, TimeUnit.SECONDS // 延长至15秒，适配设备回执场景
);
if (lockSuccess == null || !lockSuccess) {
String errorTopic = "frontend/" + clientId + "/error/" + deviceId;
mqttMessageSender.publish(errorTopic, "{\"msg\":\"设备忙，请稍后重试\"}");
String errorTopic = "frontend/" + clientId + "/dtu/" + deviceId+"/up";
mqttMessageSender.publish(errorTopic, "{\"msg\":\"设备" + funcType + "功能忙，请稍后重试\"}");
// 优化：替换System.err为log.warn
log.warn("【分布式锁】前端{}操作设备{}失败", clientId, deviceId);
log.warn("【分布式锁】前端{}操作设备{}的{}功能失败", clientId, deviceId, funcType);
return;
}
// 3. 记录日志
log.info("【指令处理】前端{}于{}控制设备{}，指令：{}",
clientId, LocalDateTime.now(), deviceId, payload);
log.info("【指令处理】前端{}于{}控制设备{}的{}功能，指令：{}",
clientId, LocalDateTime.now(), deviceId, funcType, payload);
// 4. 转发指令到设备
String deviceTopic = "dtu/" + deviceId + "/control";
String deviceTopic = "dtu/" + deviceId + "/down";
mqttMessageSender.publish(deviceTopic, payload);
// 优化：替换System.out为log.info
log.info("【指令转发】前端{} → 设备{}", clientId, deviceId);
log.info("【指令转发】前端{} → 设备{}的{}功能", clientId, deviceId, funcType);
}
/**
@ -277,11 +355,9 @@ public class MqttMessageHandler implements SmartLifecycle {
*/
private boolean checkPermission(String clientId, String deviceId) {
// 管理员权限：clientId以admin_开头
if (clientId.startsWith("admin_")) {
return true;
}
// 普通用户权限：校验Redis中是否绑定该设备：校验Redis中user_device:{clientId}是否包含该设备ID
return Boolean.TRUE.equals(stringRedisTemplate.opsForSet().isMember("user_device:" + clientId, deviceId));
return Boolean.TRUE;
}
/**
@ -296,6 +372,7 @@ public class MqttMessageHandler implements SmartLifecycle {
// 保存订阅关系到Redis
stringRedisTemplate.opsForSet().add("sub:" + deviceId, clientId);
stringRedisTemplate.opsForSet().add("subc:" + clientId, deviceId); // 新增：反向索引
// 优化：替换System.out为log.info
log.info("【订阅管理】前端{}订阅设备{}成功", clientId, deviceId);
@ -328,12 +405,14 @@ public class MqttMessageHandler implements SmartLifecycle {
// 从Redis删除订阅关系
stringRedisTemplate.opsForSet().remove("sub:" + deviceId, clientId);
stringRedisTemplate.opsForSet().remove("subc:" + clientId, deviceId); // 新增：反向索引
// 优化：替换System.out为log.info
log.info("【前端取消订阅】前端{}取消订阅设备{}成功", clientId, deviceId);
}
/**
* 批量取消前端的所有设备订阅（核心：根据clientId清理所有订阅关系）
*
* @param clientId 前端唯一标识（如wx_123）
* @return 前端需要取消的MQTT主题列表（供前端批量取消）
*/
@ -353,27 +432,23 @@ public class MqttMessageHandler implements SmartLifecycle {
}
// 步骤1：查询该前端订阅的所有设备ID（生产环境用Scan替代Keys，避免阻塞Redis）
Set<String> subKeys = scanRedisKeys("sub:*");
List<String> deviceIds = new ArrayList<>();
Set<String> deviceSet = stringRedisTemplate.opsForSet().members("subc:" + clientId);
if (deviceSet == null || deviceSet.isEmpty()) {
return Collections.emptyList();
}
List<String> frontendTopics = new ArrayList<>();
if (subKeys != null && !subKeys.isEmpty()) {
for (String subKey : subKeys) {
// 检查该sub:{deviceId}集合中是否包含当前clientId
Boolean isMember = stringRedisTemplate.opsForSet().isMember(subKey, clientId);
if (Boolean.TRUE.equals(isMember)) {
// 解析设备ID：sub:1001 → 1001
String deviceId = subKey.split(":")[1];
deviceIds.add(deviceId);
// 构建前端需要取消的MQTT主题
for (String deviceId : deviceSet) {
String subKey = "sub:" + deviceId;
stringRedisTemplate.opsForSet().remove(subKey, clientId);
String frontendTopic = "frontend/" + clientId + "/dtu/" + deviceId + "/up";
frontendTopics.add(frontendTopic);
// 从该设备的订阅列表中移除clientId
stringRedisTemplate.opsForSet().remove(subKey, clientId);
log.info("【批量取消】前端{}取消设备{}订阅", clientId, deviceId);
}
}
}
// 删除反向索引
stringRedisTemplate.delete("subc:" + clientId);
// 步骤2：清理该前端的分布式锁（可选，防止死锁）
Set<String> lockKeys = scanRedisKeys("lock:*");
@ -387,7 +462,7 @@ public class MqttMessageHandler implements SmartLifecycle {
}
}
log.info("【批量取消】前端{}共取消{}个设备订阅", clientId, deviceIds.size());
log.info("【批量取消】前端{}共取消{}个设备订阅", clientId, deviceSet.size());
return frontendTopics;
}
@ -418,10 +493,11 @@ public class MqttMessageHandler implements SmartLifecycle {
}
// ========== 手动重连接口（供Controller调用） ==========
/**
* 手动触发MQTT重连（最小改动：不替换client实例，只用同一个client重连）
*/
public String manualReconnect() {
public synchronized String manualReconnect() {
isRunning.set(true);
try {
// 强制断开旧连接（如果存在）
@ -448,6 +524,7 @@ public class MqttMessageHandler implements SmartLifecycle {
}
// ======================== SmartLifecycle 生命周期管理（核心修复） ========================
/**
* 启动MQTT客户端（Spring上下文初始化/重启时触发）
* 核心：替代@PostConstruct，保证上下文重启时重新初始化MQTT连接