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

暂提

feasure
xce 2026-01-16 01:55:40 +08:00
parent 6e9e54bc41
commit 913365927f
3 changed files with 133 additions and 288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
agri-admin/src/main/java/com/agri/web/controller/mqtt/MqttController.java
View File

@ -1,21 +1,14 @@
package com.agri.web.controller.mqtt;
import com.agri.framework.interceptor.MqttMessageHandler;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.Map;
import java.util.List;
/**
* @Auther: jone
* @Date: 2026/1/15 - 01 - 15 - 23:45
* @Description: com.agri.web.controller.mqtt
* @version: 1.0
* MQTT订阅管理接口
*/
@RestController
@RequestMapping("/api/mqtt")
public class MqttController {
@ -24,30 +17,29 @@ public class MqttController {
private MqttMessageHandler mqttMessageHandler;
/**
* 前端订阅设备状态
* 单个订阅
*/
@PostMapping("/subscribe")
public String subscribe(@RequestBody Map<String, String> params) {
String clientId = params.get("clientId");
String deviceId = params.get("deviceId");
if (clientId == null || deviceId == null) {
return "参数错误";
}
@PostMapping("/single")
public String subscribe(@RequestParam String clientId, @RequestParam String deviceId) {
mqttMessageHandler.subscribeDevice(clientId, deviceId);
return "订阅成功";
}
/**
* 前端取消订阅设备状态
* 单个取消
*/
@PostMapping("/unsubscribe")
public String unsubscribe(@RequestBody Map<String, String> params) {
String clientId = params.get("clientId");
String deviceId = params.get("deviceId");
if (clientId == null || deviceId == null) {
return "参数错误";
}
@DeleteMapping("/single")
public String unsubscribe(@RequestParam String clientId, @RequestParam String deviceId) {
mqttMessageHandler.unsubscribeDevice(clientId, deviceId);
return "取消订阅成功";
}
/**
* 批量取消当前用户的所有设备订阅
*/
@DeleteMapping("/batch")
public List<String> unsubscribeAll(@RequestParam String clientId) {
// 返回前端需要取消的MQTT主题列表
return mqttMessageHandler.unsubscribeAllDevice(clientId);
}
}

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

@ -5,6 +5,12 @@ spring:
port: 6379
password: lld123
database: 1
lettuce:
pool:
max-active: 8
max-idle: 8
min-idle: 2
max-wait: 10000ms
# MQTT配置
mqtt:
host: tcp://122.51.109.52:1883 # 设备/后端的MQTT TCP地址
@ -12,7 +18,7 @@ spring:
username: admin # Mosquitto共用账号
password: Admin#12345678 # Mosquitto密码
client-id: springboot-backend-${random.uuid} # 后端客户端ID（唯一）
default-topic: device/+/status,device/+/heartbeat,frontend/+/control/+ # 后端监听的主题
default-topic: dtu/+/up,frontend/+/control/+ # 后端监听的主题
qos: 1 # 消息可靠性
timeout: 60 # 连接超时
keep-alive: 60 # 心跳间隔

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

@ -9,27 +9,23 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* MQTT消息处理器
* MQTT消息处理器（无心跳包版本）
* 核心功能：
* 1. 订阅设备状态/心跳、前端控制指令主题
* 2. 转发设备状态到订阅的前端
* 3. 处理设备心跳，维护在线状态并上报4G平台
* 4. 处理前端控制指令，权限校验+分布式锁+转发给设备
* 5. 定时检查离线设备，更新状态并上报
*
* @Auther: lld
* @Date: 2026/1/15 - 01 - 15 - 23:43
* @version: 1.0
* 1. 订阅设备状态、前端控制指令主题
* 2. 管理前端-设备订阅关系（Redis）
* 3. 转发设备状态到订阅的前端
* 4. 处理前端控制指令（权限校验+分布式锁+转发）
* 适配JDK 8，无心跳包相关逻辑
*/
@Component
public class MqttMessageHandler {
@ -46,34 +42,22 @@ public class MqttMessageHandler {
@Resource
private StringRedisTemplate stringRedisTemplate;
/** 4G平台API地址：用于上报设备在线/离线状态 */
private static final String FOUR_G_API = "http://你的4G平台IP/api/device/status";
/** 心跳超时时间（秒）：设备超过该时间未发心跳则判定为离线 */
private static final long HEARTBEAT_TIMEOUT = 60;
@Value("${spring.mqtt.default-topic}")
// 读取配置文件中的默认订阅主题（移除心跳主题）
@Value("${spring.mqtt.default-topic:device/+/status,frontend/+/control/+}")
private String defaultTopic;
/**
* 初始化方法：项目启动时执行
* 1. 设置MQTT回调函数
* 2. 订阅核心主题
* 3. 启动离线设备检查线程
*
* @throws MqttException MQTT订阅失败异常
* 初始化：订阅主题+设置回调
*/
@PostConstruct
public void subscribeTopics() throws MqttException {
// 定义需要监听的MQTT主题数组
// device/+/status：所有设备的业务状态（温湿度、开关等）
// device/+/heartbeat：所有设备的心跳包（用于判定在线状态）
// frontend/+/control/+：所有前端发送的设备控制指令
// 解析配置文件中的主题列表（逗号分隔）
// 解析配置的主题列表
String[] topics = defaultTopic.split(",");
// 对应主题的QoS级别（所有主题使用相同QoS，也可自定义多QoS配置）
int[] qos = new int[topics.length];
// 所有主题QoS=1
Arrays.fill(qos, 1);
int[] qosArray = new int[topics.length];
// 按主题类型设置QoS：控制指令/状态用QoS 1
for (int i = 0; i < topics.length; i++) {
qosArray[i] = 1;
}
// 设置MQTT消息回调：处理连接断开、消息接收、消息发布完成
mqttClient.setCallback(new MqttCallback() {
@ -104,44 +88,30 @@ public class MqttMessageHandler {
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// 可添加消息发布成功的日志
// System.out.println("消息发布完成：" + token.getTopics()[0]);
// QoS 1确认：消息已被服务端接收
System.out.println("【MQTT确认】消息发布完成，主题：" + token.getTopics()[0]);
}
});
// 订阅主题：阻塞式操作，订阅成功后继续执行
mqttClient.subscribe(topics, qos);
System.out.println("【MQTT初始化】核心主题订阅成功，订阅列表：" + String.join(",", topics));
// 启动离线设备检查线程：独立线程，避免阻塞主线程
// 线程名：offline-check-thread，便于日志排查
new Thread(new Runnable() {
@Override
public void run() {
checkOfflineDevice();
}
}, "offline-check-thread").start();
System.out.println("【MQTT初始化】离线设备检查线程已启动");
// 订阅主题
mqttClient.subscribe(topics, qosArray);
System.out.println("【MQTT初始化】订阅主题：" + String.join(",", topics));
}
/**
* 消息分发处理：根据主题类型路由到不同处理方法
* 消息分发处理：根据主题类型路由到不同处理方法\仅处理设备状态、前端控制指令
* @param topic 消息主题
* @param payload 消息内容（JSON字符串）
*/
private void handleMessage(String topic, String payload) {
try {
System.out.println("【MQTT消息接收】topic=" + topic + ", payload=" + payload);
System.out.println("【MQTT接收】topic=" + topic + ", payload=" + payload);
// 1. 处理设备业务状态主题：device/{deviceId}/status
// 设备状态主题：device/{deviceId}/status
if (topic.matches("device/\\w+/status")) {
handleDeviceStatus(topic, payload);
}
// 2. 处理设备心跳主题：device/{deviceId}/heartbeat
else if (topic.matches("device/\\w+/heartbeat")) {
handleDeviceHeartbeat(topic, payload);
}
// 3. 处理前端控制指令主题：frontend/{clientId}/control/{deviceId}
// 处理前端控制指令主题：frontend/{clientId}/control/{deviceId}
else if (topic.matches("frontend/\\w+/control/\\w+")) {
handleFrontendControl(topic, payload);
}
@ -152,38 +122,21 @@ public class MqttMessageHandler {
}
/**
* 处理设备业务状态消息
* 逻辑：
* 1. 解析设备ID
* 2. 补充设备在线状态到消息体
* 3. 查询订阅该设备的前端列表
* 4. 转发消息到每个前端的专属主题
*
* @param topic 消息主题（device/{deviceId}/status）
* @param payload 设备状态JSON字符串
* @throws MqttException 消息发布异常
* 处理设备状态：转发给订阅的前端
*/
private void handleDeviceStatus(String topic, String payload) throws MqttException {
// 解析设备ID：主题格式为device/{deviceId}/status，分割后第2个元素是设备ID
String deviceId = topic.split("/")[1];
// 补充设备在线状态到payload（兼容JSON格式）
// 从Redis获取设备在线状态：device:online:{deviceId} → true/false
String onlineStatus = stringRedisTemplate.opsForValue().get("device:online:" + deviceId);
// 若Redis中无记录，默认离线
String finalOnlineStatus = (onlineStatus == null) ? "false" : onlineStatus;
// 拼接在线状态到JSON末尾（兼容无空格的JSON格式）
String newPayload = payload.replace("}", ",\"online\":\"" + finalOnlineStatus + "\"}");
// 查询Redis中订阅该设备的前端clientId列表：sub:{deviceId} → Set<String>
// 查询Redis中订阅该设备的前端列表：sub:{deviceId}
Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);
if (subscribedClients != null && !subscribedClients.isEmpty()) {
// 遍历所有订阅的前端，推送消息到前端专属主题
// 推送给每个订阅的前端
for (String clientId : subscribedClients) {
// 前端专属主题：frontend/{clientId}/device/{deviceId}/status
String frontendTopic = "frontend/" + clientId + "/device/" + deviceId + "/status";
// 发布消息（保留最新消息，前端订阅后可立即获取）
mqttMessageSender.publish(frontendTopic, newPayload);
// 发布消息
mqttMessageSender.publish(frontendTopic, payload);
System.out.println("【设备状态转发】设备" + deviceId + " → 前端" + clientId + "，主题：" + frontendTopic);
}
} else {
@ -192,109 +145,44 @@ public class MqttMessageHandler {
}
/**
* 处理设备心跳消息
* 逻辑：
* 1. 解析设备ID
* 2. 更新Redis中设备最后心跳时间、在线状态
* 3. 异步上报在线状态到4G平台（避免阻塞MQTT消息处理）
*
* @param topic 消息主题（device/{deviceId}/heartbeat）
* @param payload 心跳包JSON字符串（可包含timestamp等字段）
*/
private void handleDeviceHeartbeat(String topic, String payload) {
// 解析设备ID：主题格式为device/{deviceId}/heartbeat，分割后第2个元素是设备ID
String deviceId = topic.split("/")[1];
// 获取当前时间戳（秒）
long currentTime = System.currentTimeMillis() / 1000;
// 更新Redis：存储最后心跳时间 → device:last_heartbeat:{deviceId}
stringRedisTemplate.opsForValue().set("device:last_heartbeat:" + deviceId, String.valueOf(currentTime));
// 更新Redis：存储在线状态，设置过期时间（心跳超时+10秒），避免Redis数据堆积
stringRedisTemplate.opsForValue().set(
"device:online:" + deviceId,
"true",
HEARTBEAT_TIMEOUT + 10,
TimeUnit.SECONDS
);
// 异步上报4G平台：使用独立线程，避免阻塞MQTT消息处理线程
new Thread(new Runnable() {
@Override
public void run() {
try {
// 构造上报4G平台的JSON数据
String statusJson = String.format(
"{\"device_id\":\"%s\",\"online\":true,\"timestamp\":%d}",
deviceId, currentTime
);
// 调用4G平台API（POST请求）
RestTemplate restTemplate = new RestTemplate();
String response = restTemplate.postForObject(FOUR_G_API, statusJson, String.class);
System.out.println("【4G平台上报】设备" + deviceId + "在线状态上报成功，响应：" + response);
} catch (Exception e) {
System.err.println("【4G平台上报】设备" + deviceId + "在线状态上报失败，异常：" + e.getMessage());
}
}
}).start();
}
/**
* 处理前端控制指令
* 逻辑：
* 1. 解析前端clientId、设备ID
* 2. 权限校验：验证前端是否有权操作该设备
* 3. 分布式锁：避免多前端同时控制同一设备
* 4. 记录操作日志
* 5. 转发指令到设备专属主题
*
* @param topic 消息主题（frontend/{clientId}/control/{deviceId}）
* @param payload 控制指令JSON字符串
* @throws MqttException 消息发布异常
* 处理前端控制指令：权限校验+分布式锁+转发给设备
*/
private void handleFrontendControl(String topic, String payload) throws MqttException {
// 解析主题：frontend/{clientId}/control/{deviceId}
// 解析前端clientId、设备ID
String[] parts = topic.split("/");
String clientId = parts[1]; // 前端唯一标识
String deviceId = parts[3]; // 目标设备ID
String clientId = parts[1];
String deviceId = parts[3];
// 1. 权限校验：失败则推送错误消息给前端
// 1. 权限校验（示例：admin开头有全权限）
if (!checkPermission(clientId, deviceId)) {
String errorTopic = "frontend/" + clientId + "/error/" + deviceId;
mqttMessageSender.publish(errorTopic, "{\"msg\":\"无设备操作权限\"}");
System.err.println("【前端指令处理】前端" + clientId + "操作设备" + deviceId + "权限校验失败");
System.err.println("【权限校验】前端" + clientId + "操作设备" + deviceId + "失败");
return;
}
// 2. 分布式锁：lock:{deviceId}，过期时间10秒（避免死锁）
// 2. 分布式锁：避免多前端并发控制
String lockKey = "lock:" + deviceId;
Boolean lockSuccess = stringRedisTemplate.opsForValue().setIfAbsent(
lockKey,
clientId,
10,
TimeUnit.SECONDS
lockKey, clientId, 10, TimeUnit.SECONDS
);
// 锁获取失败：设备忙，推送错误消息给前端
if (lockSuccess == null || !lockSuccess) {
String errorTopic = "frontend/" + clientId + "/error/" + deviceId;
mqttMessageSender.publish(errorTopic, "{\"msg\":\"设备忙，请稍后重试\"}");
System.err.println("【前端指令处理】前端" + clientId + "操作设备" + deviceId + "获取锁失败（设备忙）");
System.err.println("【分布式锁】前端" + clientId + "操作设备" + deviceId + "失败");
return;
}
// 3. 记录操作日志（示例：可替换为数据库存储）
// 3. 记录日志
System.out.println(String.format(
"【前端指令处理】前端%s于%s控制设备%s，指令：%s",
"【指令处理】前端%s于%s控制设备%s，指令：%s",
clientId, LocalDateTime.now(), deviceId, payload
));
// 4. 转发指令到设备专属主题：device/{deviceId}/control
// 4. 转发指令到设备
String deviceTopic = "device/" + deviceId + "/control";
mqttMessageSender.publish(deviceTopic, payload);
System.out.println("【前端指令转发】前端" + clientId + " → 设备" + deviceId + "，主题：" + deviceTopic);
System.out.println("【指令转发】前端" + clientId + " → 设备" + deviceId);
}
/**
@ -312,117 +200,27 @@ public class MqttMessageHandler {
if (clientId.startsWith("admin_")) {
return true;
}
// 普通用户权限：校验Redis中是否绑定该设备
return Boolean.TRUE.equals(stringRedisTemplate.opsForSet().isMember("user_device:" + clientId, deviceId));
// 普通用户权限：校验Redis中是否绑定该设备：校验Redis中user_device:{clientId}是否包含该设备ID
return stringRedisTemplate.opsForSet().isMember("user_device:" + clientId, deviceId);
}
/**
* 定时检查离线设备（无限循环，每10秒执行一次）
* 逻辑：
* 1. 获取所有设备的最后心跳记录
* 2. 判定是否离线（当前时间 - 最后心跳时间 > 心跳超时）
* 3. 更新Redis在线状态为离线
* 4. 上报离线状态到4G平台
*/
private void checkOfflineDevice() {
while (true) {
try {
// 获取Redis中所有设备的最后心跳记录：device:last_heartbeat:*
Set<String> heartbeatKeys = stringRedisTemplate.keys("device:last_heartbeat:*");
// 无设备心跳记录，休眠10秒后继续
if (heartbeatKeys == null || heartbeatKeys.isEmpty()) {
Thread.sleep(10000);
continue;
}
// 当前时间戳（秒）
long currentTime = System.currentTimeMillis() / 1000;
// 遍历所有设备心跳记录
for (String key : heartbeatKeys) {
// 解析设备ID：key格式为device:last_heartbeat:{deviceId}
String deviceId = key.split(":")[2];
// 获取最后心跳时间
String lastHeartbeatStr = stringRedisTemplate.opsForValue().get(key);
// 无心跳记录，跳过
if (lastHeartbeatStr == null) {
continue;
}
// 转换为长整型
long lastHeartbeat = Long.parseLong(lastHeartbeatStr);
// 判定离线：超过心跳超时时间未发心跳
if (currentTime - lastHeartbeat > HEARTBEAT_TIMEOUT) {
// 更新Redis在线状态为离线
stringRedisTemplate.opsForValue().set("device:online:" + deviceId, "false");
// 构造上报4G平台的JSON数据
String statusJson = String.format(
"{\"device_id\":\"%s\",\"online\":false,\"timestamp\":%d}",
deviceId, currentTime
);
// 调用4G平台API上报离线状态
RestTemplate restTemplate = new RestTemplate();
restTemplate.postForObject(FOUR_G_API, statusJson, String.class);
System.out.println("【离线设备检查】设备" + deviceId + "判定为离线，已上报4G平台");
}
}
// 每10秒检查一次
Thread.sleep(10000);
} catch (InterruptedException e) {
// 线程中断，退出循环
System.err.println("【离线设备检查】线程被中断，停止检查：" + e.getMessage());
Thread.currentThread().interrupt();
break;
} catch (Exception e) {
System.err.println("【离线设备检查】异常：" + e.getMessage());
// 异常时休眠10秒，避免无限循环报错
try {
Thread.sleep(10000);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
}
/**
* 前端订阅设备状态接口（供Controller层调用）
* 逻辑：
* 1. 将前端clientId加入设备的订阅列表
* 2. 推送设备最新状态给前端（立即）
*
* @param clientId 前端唯一标识
* @param deviceId 设备ID
* 前端订阅设备（Controller调用）
*/
public void subscribeDevice(String clientId, String deviceId) {
// 将前端clientId添加到设备的订阅列表：sub:{deviceId}
// 保存订阅关系到Redis
stringRedisTemplate.opsForSet().add("sub:" + deviceId, clientId);
System.out.println("【前端订阅】前端" + clientId + "订阅设备" + deviceId + "成功");
System.out.println("【订阅管理】前端" + clientId + "订阅设备" + deviceId + "成功");
// 推送设备最新状态给前端（立即）
try {
// 从Redis获取设备最新状态：device:latest:{deviceId}
String latestStatus = stringRedisTemplate.opsForValue().get("device:latest:" + deviceId);
if (latestStatus != null) {
// 前端专属主题
String frontendTopic = "frontend/" + clientId + "/device/" + deviceId + "/status";
// 推送设备最新状态（可选）
String latestStatus = stringRedisTemplate.opsForValue().get("device:latest:" + deviceId);
if (latestStatus != null) {
String frontendTopic = "frontend/" + clientId + "/device/" + deviceId + "/status";
try {
mqttMessageSender.publish(frontendTopic, latestStatus);
System.out.println("【前端订阅】推送设备" + deviceId + "最新状态给前端" + clientId);
} else {
System.out.println("【前端订阅】设备" + deviceId + "无最新状态，跳过推送");
} catch (MqttException e) {
System.err.println("【订阅推送】设备" + deviceId + "状态推送失败：" + e.getMessage());
}
} catch (MqttException e) {
System.err.println("【前端订阅】推送设备" + deviceId + "状态给前端" + clientId + "失败：" + e.getMessage());
e.printStackTrace();
}
}
@ -434,8 +232,57 @@ public class MqttMessageHandler {
* @param deviceId 设备ID
*/
public void unsubscribeDevice(String clientId, String deviceId) {
// 从设备订阅列表移除前端clientId
// 从Redis删除订阅关系
stringRedisTemplate.opsForSet().remove("sub:" + deviceId, clientId);
System.out.println("【前端取消订阅】前端" + clientId + "取消订阅设备" + deviceId + "成功");
}
/**
* 批量取消前端的所有设备订阅（核心：根据clientId清理所有订阅关系）
* @param clientId 前端唯一标识（如wx_123）
* @return 前端需要取消的MQTT主题列表（供前端批量取消）
*/
public List<String> unsubscribeAllDevice(String clientId) {
// 步骤1：查询该前端订阅的所有设备ID（Redis中所有sub:*集合中包含该clientId的key）
// 注意：生产环境建议用scan代替keys，避免阻塞Redis
Set<String> subKeys = stringRedisTemplate.keys("sub:*");
List<String> deviceIds = new ArrayList<>();
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主题
String frontendTopic = "frontend/" + clientId + "/device/" + deviceId + "/status";
frontendTopics.add(frontendTopic);
// 从该设备的订阅列表中移除clientId
stringRedisTemplate.opsForSet().remove(subKey, clientId);
System.out.println("【批量取消】前端" + clientId + "取消设备" + deviceId + "订阅");
}
}
}
// 步骤2：清理该前端的分布式锁（可选，防止死锁）
Set<String> lockKeys = stringRedisTemplate.keys("lock:*");
if (lockKeys != null && !lockKeys.isEmpty()) {
for (String lockKey : lockKeys) {
String lockValue = stringRedisTemplate.opsForValue().get(lockKey);
if (clientId.equals(lockValue)) {
stringRedisTemplate.delete(lockKey);
System.out.println("【批量取消】清理前端" + clientId + "持有的锁：" + lockKey);
}
}
}
System.out.println("【批量取消】前端" + clientId + "共取消" + deviceIds.size() + "个设备订阅");
return frontendTopics;
}
}