单线程

agri-admin/src/main/resources/application-mqtt.yml

@@ -5,8 +5,10 @@ spring:
     username: admin  # Mosquitto共用账号
     password: Admin#12345678 # Mosquitto密码
     client-id: springboot-backend  # 截取UUID前8位（自动去横线）
-    default-topic: dtu/+/up,frontend/+/control/+ # 后端监听的主题
+    default-topic: dtu/+/up,frontend/+/control/+,frontend/+/online # 后端监听的主题
     qos: 1  # 消息可靠性
     timeout: 60 # 连接超时
     keep-alive: 60 # 心跳间隔
+    auto-off-seconds: 30 #自动关延迟秒数。
+    latest-ttl-seconds: 120 #设备最新状态缓存的过期时间（秒）。
     

agri-framework/src/main/java/com/agri/framework/interceptor/MqttMessageHandler.java

@@ -1,6 +1,8 @@
 package com.agri.framework.interceptor;
 
 import com.agri.framework.config.MqttConfig;
+import com.agri.system.domain.SysAgriLimit;
+import com.agri.system.service.ISysAgriLimitService;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
 import com.alibaba.fastjson2.TypeReference;
@@ -12,6 +14,7 @@ import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -26,12 +29,18 @@ import java.nio.charset.StandardCharsets;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 
 /**
  * MQTT消息处理器（无心跳包版本）
@@ -78,6 +87,14 @@ public class MqttMessageHandler implements SmartLifecycle {
     @Value("${spring.mqtt.default-topic}")
     private String defaultTopic;
 
+    // 新增：自动关延迟秒数（固定多少秒-n秒）
+    @Value("${spring.mqtt.auto-off-seconds:30}")
+    private int autoOffSeconds;
+
+    // 新增：最新状态缓存TTL（设备每10秒上报一次，缓存一小段时间即可）
+    @Value("${spring.mqtt.latest-ttl-seconds:120}")
+    private int latestTtlSeconds;
+
     // 优化：统一使用SLF4J日志（JDK 8兼容）
     private static final Logger log = LoggerFactory.getLogger(MqttMessageHandler.class);
 
@@ -90,6 +107,29 @@ public class MqttMessageHandler implements SmartLifecycle {
     @Resource
     private MqttConnectOptions mqttConnectOptions;
 
+    // 新增：自动关任务线程池（单线程，避免并发执行）
+    private final ScheduledExecutorService autoOffExecutor = Executors.newSingleThreadScheduledExecutor();
+
+    // 新增：同设备同功能只保留最后一次自动关任务
+    private final ConcurrentHashMap<String, ScheduledFuture<?>> autoOffFutureMap = new ConcurrentHashMap<>();
+
+    @Autowired
+    private ISysAgriLimitService agriLimitService;
+
+
+    // 初始化映射（建议放在类初始化块/构造方法中，只初始化一次）
+    private static final Map<String, Function<SysAgriLimit, Integer>> LIMIT_MAP = new HashMap<>();
+    static {
+        LIMIT_MAP.put("jm1g", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm1gLimit())));
+        LIMIT_MAP.put("jm2g", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm2gLimit())));
+        LIMIT_MAP.put("jbg", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJbgLimit())));
+        LIMIT_MAP.put("jm3g", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm3gLimit())));
+        LIMIT_MAP.put("jm2k", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm2kLimit())));
+        LIMIT_MAP.put("jm3k", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm3kLimit())));
+        LIMIT_MAP.put("jbk", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJbkLimit())));
+        LIMIT_MAP.put("jm1k", agriLimit -> Integer.parseInt(String.valueOf(agriLimit.getJm1kLimit())));
+    }
+
     /**
      * 初始化：订阅主题+设置回调
      * （移除@PostConstruct，改为由SmartLifecycle的start()触发）
@@ -234,12 +274,21 @@ public class MqttMessageHandler implements SmartLifecycle {
 
         // 第二步：判断是否为设备回执（{"suc":true/false,"prop":{"功能码":指令}}）
         String funcType = null;
+        Integer funcValue = null;
+        boolean isAck = false;
+
         if (payloadObj.containsKey("suc") && payloadObj.containsKey("prop")) {
+            isAck = true;
             JSONObject propObj = payloadObj.getJSONObject("prop");
             if (propObj != null && !propObj.isEmpty()) {
                 // 提取prop中的第一个功能码
                 Map.Entry<String, Object> propEntry = propObj.entrySet().iterator().next();
                 funcType = propEntry.getKey();
+                try {
+                    funcValue = Integer.parseInt(String.valueOf(propEntry.getValue()));
+                } catch (Exception ignore) {
+                }
+
                 // 释放对应功能的分布式锁
                 String lockKey = "lock:" + deviceId + ":" + funcType;
                 Boolean delete = stringRedisTemplate.delete(lockKey);
@@ -248,6 +297,23 @@ public class MqttMessageHandler implements SmartLifecycle {
                 }
                 log.info("【设备回执】设备{}的{}功能执行完成，已释放锁：{},{}", deviceId, funcType, lockKey, delete);
 
+                // 新增：回执固定是{"suc":true,"prop":{"jm1k":1}}
+                boolean suc = payloadObj.getBooleanValue("suc");
+                if (suc && StringUtils.hasText(funcType) && funcValue != null && funcValue == 1) {
+
+                    SysAgriLimit agriLimit = agriLimitService.lambdaQuery()
+                            .eq(SysAgriLimit::getImei, deviceId)
+                            .one();
+                    int autoOffSeconds = 0;
+
+                    if (agriLimit != null) {
+                        autoOffSeconds = LIMIT_MAP.getOrDefault(funcType, k -> 0).apply(agriLimit);
+                    }
+
+                    // 新增：回执成功且值=1（表示运行/开启）时，起个任务，固定多少秒-n秒
+                    scheduleAutoOff(deviceId, funcType, autoOffSeconds);
+                }
+
                 // 广播回执结果给所有订阅该设备的前端
                 // String broadcastTopic =  "frontend/" + clientId + "/dtu/" + deviceId + "/listener";;
                 // JSONObject ackPayload = new JSONObject();
@@ -259,6 +325,26 @@ public class MqttMessageHandler implements SmartLifecycle {
             }
         }
 
+        // 新增：设备每10秒上报一次状态包，写入latest供自动关任务读取（只在需要时写，减少消耗）
+        if (!isAck) {
+            boolean needWriteLatest = false;
+
+            // 1) 有人订阅时才写latest
+            Set<String> subscribedClients = stringRedisTemplate.opsForSet().members("sub:" + deviceId);
+            if (subscribedClients != null && !subscribedClients.isEmpty()) {
+                needWriteLatest = true;
+            }
+
+            // 2) 或者存在该设备的自动关任务时也写latest（任务需要最新状态判断）
+            if (!needWriteLatest && hasAutoOffTask(deviceId)) {
+                needWriteLatest = true;
+            }
+
+            if (needWriteLatest) {
+                stringRedisTemplate.opsForValue().set("device:latest:" + deviceId, payload, latestTtlSeconds, TimeUnit.SECONDS);
+            }
+        }
+
         // 非回执消息：正常转发给订阅前端
         // if (!isDeviceAck) {
         // 查询Redis中订阅该设备的前端列表：sub:{deviceId}
@@ -281,6 +367,90 @@ public class MqttMessageHandler implements SmartLifecycle {
         // }
     }
 
+    // 新增：是否存在该设备的自动关任务
+    private boolean hasAutoOffTask(String deviceId) {
+        if (!StringUtils.hasText(deviceId)) {
+            return false;
+        }
+        String prefix = "autooff:" + deviceId + ":";
+        for (String key : autoOffFutureMap.keySet()) {
+            if (key != null && key.startsWith(prefix)) {
+                ScheduledFuture<?> f = autoOffFutureMap.get(key);
+                if (f != null && !f.isCancelled() && !f.isDone()) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    // 新增：起个任务，固定多少秒-n秒，【监听最新的设备状态，如果还在运行】，发送设备关的指令
+    private void scheduleAutoOff(String deviceId, String funcType, int delaySeconds) {
+        if (!StringUtils.hasText(deviceId) || !StringUtils.hasText(funcType) || delaySeconds <= 0) {
+            return;
+        }
+        String taskKey = "autooff:" + deviceId + ":" + funcType;
+
+        // 同设备同功能只保留最后一次
+        ScheduledFuture<?> old = autoOffFutureMap.remove(taskKey);
+        if (old != null) {
+            old.cancel(false);
+        }
+
+        ScheduledFuture<?> future = autoOffExecutor.schedule(() -> {
+            try {
+                runAutoOff(deviceId, funcType);
+            } catch (Exception e) {
+                log.error("【自动关任务】执行失败，deviceId={}, funcType={}", deviceId, funcType, e);
+            } finally {
+                autoOffFutureMap.remove(taskKey);
+            }
+        }, delaySeconds, TimeUnit.SECONDS);
+
+        autoOffFutureMap.put(taskKey, future);
+        log.info("【自动关任务】已创建：deviceId={}, funcType={}, delay={}s", deviceId, funcType, delaySeconds);
+    }
+
+    // 新增：读取最新状态（device:latest:{deviceId}），若仍为1则下发 {"funcType":0} 到 dtu/{id}/down
+    private void runAutoOff(String deviceId, String funcType) throws MqttException {
+        String latest = stringRedisTemplate.opsForValue().get("device:latest:" + deviceId);
+        if (!StringUtils.hasText(latest)) {
+            log.warn("【自动关任务】无最新状态，跳过：deviceId={}, funcType={}", deviceId, funcType);
+            return;
+        }
+
+        JSONObject latestObj;
+        try {
+            latestObj = JSON.parseObject(latest);
+        } catch (Exception e) {
+            log.warn("【自动关任务】最新状态JSON解析失败，跳过：deviceId={}, funcType={}", deviceId, funcType);
+            return;
+        }
+        if (latestObj == null || latestObj.isEmpty()) {
+            return;
+        }
+
+        // 设备每10秒上报的状态包：{"jm1k":0/1,...} 顶层字段直接取
+        Integer current = null;
+        try {
+            if (latestObj.containsKey(funcType)) {
+                current = latestObj.getIntValue(funcType);
+            }
+        } catch (Exception ignore) {
+        }
+
+        if (current != null && current == 1) {
+            JSONObject down = new JSONObject();
+            down.put(funcType, 0);
+
+            String deviceTopic = "dtu/" + deviceId + "/down";
+            mqttMessageSender.publish(deviceTopic, down.toJSONString());
+            log.info("【自动关任务】检测仍在运行，已下发关闭：deviceId={}, funcType={}, payload={}", deviceId, funcType, down.toJSONString());
+        } else {
+            log.info("【自动关任务】检测未运行或状态未知，跳过关闭：deviceId={}, funcType={}, current={}", deviceId, funcType, current);
+        }
+    }
+
     /**
      * 处理前端控制指令：权限校验+分布式锁+转发给设备
      */
@@ -504,7 +674,7 @@ public class MqttMessageHandler implements SmartLifecycle {
     /**
      * 手动触发MQTT重连（最小改动：不替换client实例，只用同一个client重连）
      */
-    public synchronized  String manualReconnect() {
+    public synchronized String manualReconnect() {
         isRunning.set(true);
         try {
             // 强制断开旧连接（如果存在）
@@ -560,6 +730,17 @@ public class MqttMessageHandler implements SmartLifecycle {
         // 修复：JDK 8正确的compareAndSet写法（无命名参数）
         if (isRunning.compareAndSet(true, false)) {
             try {
+                // 新增：关闭自动关任务线程池，避免线程泄漏
+                autoOffExecutor.shutdown();
+                try {
+                    if (!autoOffExecutor.awaitTermination(3, TimeUnit.SECONDS)) {
+                        autoOffExecutor.shutdownNow();
+                    }
+                } catch (InterruptedException ignore) {
+                    autoOffExecutor.shutdownNow();
+                    Thread.currentThread().interrupt();
+                }
+
                 if (mqttClient != null) {
                     // 注意：disconnect 只在已连接时调用；close 尽量无条件释放资源
                     if (mqttClient.isConnected()) {