mqtt基础版

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

@@ -1,7 +1,15 @@
 package com.agri.web.controller.mqtt;
 
 import com.agri.framework.interceptor.MqttMessageHandler;
-import org.springframework.web.bind.annotation.*;
+import com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.util.List;
@@ -13,6 +21,8 @@ import java.util.List;
 @RequestMapping("/api/mqtt")
 public class MqttController {
 
+    private static final Logger log = LoggerFactory.getLogger(MqttController.class);
+
     @Resource
     private MqttMessageHandler mqttMessageHandler;
 
@@ -21,8 +31,16 @@ public class MqttController {
      */
     @PostMapping("/single")
     public String subscribe(@RequestParam String clientId, @RequestParam String deviceId) {
+        try {
             mqttMessageHandler.subscribeDevice(clientId, deviceId);
             return "订阅成功";
+        } catch (IllegalArgumentException e) {
+            log.error("MQTT单个订阅失败：{}", e.getMessage());
+            return "订阅失败：" + e.getMessage();
+        } catch (Exception e) {
+            log.error("MQTT单个订阅异常", e);
+            return "订阅失败：系统异常";
+        }
     }
 
     /**
@@ -30,8 +48,16 @@ public class MqttController {
      */
     @DeleteMapping("/single")
     public String unsubscribe(@RequestParam String clientId, @RequestParam String deviceId) {
+        try {
             mqttMessageHandler.unsubscribeDevice(clientId, deviceId);
             return "取消订阅成功";
+        } catch (IllegalArgumentException e) {
+            log.error("MQTT单个取消订阅失败：{}", e.getMessage());
+            return "取消订阅失败：" + e.getMessage();
+        } catch (Exception e) {
+            log.error("MQTT单个取消订阅异常", e);
+            return "取消订阅失败：系统异常";
+        }
     }
 
     /**
@@ -39,7 +65,44 @@ public class MqttController {
      */
     @DeleteMapping("/batch")
     public List<String> unsubscribeAll(@RequestParam String clientId) {
+        try {
             // 返回前端需要取消的MQTT主题列表
             return mqttMessageHandler.unsubscribeAllDevice(clientId);
+        } catch (IllegalArgumentException e) {
+            log.error("MQTT批量取消订阅失败：{}", e.getMessage());
+            // 异常时返回空列表，避免前端解析失败
+            return Lists.newArrayList();
+        } catch (Exception e) {
+            log.error("MQTT批量取消订阅异常", e);
+            return Lists.newArrayList();
+        }
+    }
+
+    /**
+     * 手动触发MQTT重连
+     * 场景：配置修正后，手动恢复连接（无需重启服务）
+     */
+    @GetMapping("/reconnect")
+    public String manualReconnect() {
+        try {
+            return mqttMessageHandler.manualReconnect();
+        } catch (Exception e) {
+            log.error("MQTT手动重连异常", e);
+            return "手动重连失败：" + e.getMessage();
+        }
+    }
+
+    /**
+     * 查询当前MQTT连接状态
+     * 便于排查连接问题
+     */
+    @GetMapping("/status")
+    public String getMqttStatus() {
+        try {
+            return mqttMessageHandler.getMqttStatus();
+        } catch (Exception e) {
+            log.error("查询MQTT连接状态异常", e);
+            return "查询状态失败：" + e.getMessage();
+        }
     }
 }

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

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

agri-common/src/main/java/com/agri/common/utils/uuid/UUID.java

@@ -1,11 +1,12 @@
 package com.agri.common.utils.uuid;
 
+import com.agri.common.exception.UtilException;
+
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
-import com.agri.common.exception.UtilException;
 
 /**
  * 提供通用唯一识别码（universally unique identifier）（UUID）实现
@@ -481,4 +482,21 @@ public final class UUID implements java.io.Serializable, Comparable<UUID>
     {
         return ThreadLocalRandom.current();
     }
+
+
+    /**
+     * 生成指定长度的字母数字随机串（对齐Spring ${random.alphanumeric:n}语法）
+     * 字符集：0-9 + a-z + A-Z，和Spring配置的随机串规则一致
+     */
+    public static String generateAlphanumericRandom(int length) {
+        // 定义字母数字混合的字符集（和Spring ${random.alphanumeric}的字符集一致）
+        String chars = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
+        StringBuilder sb = new StringBuilder();
+        // 基于JVM的随机数生成，保证随机性
+        for (int i = 0; i < length; i++) {
+            int randomIndex = (int) (Math.random() * chars.length());
+            sb.append(chars.charAt(randomIndex));
+        }
+        return sb.toString();
+    }
 }

agri-framework/src/main/java/com/agri/framework/config/MqttConfig.java

@@ -1,5 +1,6 @@
 package com.agri.framework.config;
 
+import com.agri.common.utils.uuid.UUID;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -14,7 +15,6 @@ import org.springframework.util.StringUtils;
 
 import javax.annotation.PreDestroy;
 import java.nio.charset.StandardCharsets;
-import java.util.UUID;
 
 /**
  * MQTT核心配置类
@@ -92,7 +92,7 @@ public class MqttConfig {
         MemoryPersistence persistence = new MemoryPersistence();
 
         // 新增：客户端ID拼接随机后缀，避免多实例部署时冲突（生产环境必备）
-        String uniqueClientId = clientId + "_" + UUID.randomUUID().toString().substring(0, 8);
+        String uniqueClientId = clientId + "_" + UUID.generateAlphanumericRandom(8);
         MqttClient mqttClient = new MqttClient(host, uniqueClientId, persistence);
 
         // 3. 建立MQTT连接
@@ -228,4 +228,13 @@ public class MqttConfig {
             }
         }
     }
+
+
+    /**
+     * 暴露MQTT连接配置项为Bean，供其他类注入使用
+     */
+    @Bean
+    public MqttConnectOptions mqttConnectOptions() {
+        return getMqttConnectOptions();
+    }
 }

agri-framework/src/main/java/com/agri/framework/config/SecurityConfig.java

@@ -111,7 +111,7 @@ public class SecurityConfig
             .authorizeHttpRequests((requests) -> {
                 permitAllUrl.getUrls().forEach(url -> requests.antMatchers(url).permitAll());
                 // 对于登录login 注册register 验证码captchaImage 允许匿名访问
-                requests.antMatchers("/login", "/register", "/captchaImage").permitAll()
+                requests.antMatchers("/login", "/register", "/captchaImage","/api/mqtt/status").permitAll()
                     // 静态资源，可匿名访问
                     .antMatchers(HttpMethod.GET, "/", "/*.html", "/**/*.html", "/**/*.css", "/**/*.js", "/profile/**").permitAll()
                     .antMatchers("/swagger-ui.html", "/swagger-resources/**", "/webjars/**", "/*/api-docs", "/druid/**").permitAll()

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

@@ -1,11 +1,14 @@
 package com.agri.framework.interceptor;
 
+import com.agri.common.utils.uuid.UUID;
 import com.agri.framework.config.MqttConfig;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -25,6 +28,8 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -62,11 +67,48 @@ public class MqttMessageHandler implements SmartLifecycle {
     // 新增：生命周期管理标识，控制MQTT客户端启动/关闭
     private final AtomicBoolean isRunning = new AtomicBoolean(false);
 
+
+    /** MQTT连接配置项（从MqttConfig注入） */
+    @Resource
+    private MqttConnectOptions mqttConnectOptions;
+
+    // ========== 新增：重连相关配置 ==========
+    // 重连间隔（秒），可配置化
+    @Value("${spring.mqtt.reconnect-interval:5}")
+    private int reconnectInterval;
+    // 最大重连次数（-1表示无限重连）
+    @Value("${spring.mqtt.max-reconnect-times:-1}")
+    private int maxReconnectTimes;
+    // 重连线程池（单线程，避免并发重连）
+    private final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
+    // 当前重连次数计数
+    private int currentReconnectCount = 0;
+    // 配置错误标记（避免配置错时无限重连）
+    private volatile boolean isConfigError = false;
+
     /**
      * 初始化：订阅主题+设置回调
      * （移除@PostConstruct，改为由SmartLifecycle的start()触发）
      */
     public void subscribeTopics() throws MqttException {
+        // 关键补充1：判空（重连后替换的新实例可能为空）
+        if (mqttClient == null) {
+            log.error("【MQTT初始化】客户端实例为空，无法订阅主题");
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+        }
+
+        // 关键补充：先确保客户端已连接（如果没连则连接，已连则跳过）
+        if (!mqttClient.isConnected()) {
+            try {
+                // 使用注入的连接配置项连接Broker（带用户名密码、重连等配置）
+                mqttClient.connect(mqttConnectOptions);
+                log.info("【MQTT连接】客户端已成功连接到Broker，clientId：{}", mqttClient.getClientId());
+            } catch (MqttException e) {
+                log.error("【MQTT连接】连接Broker失败，clientId：{}", mqttClient.getClientId(), e);
+                throw e; // 抛出异常，让start()处理重连
+            }
+        }
+
         // 解析配置的主题列表
         String[] topics = defaultTopic.split(",");
         int[] qosArray = new int[topics.length];
@@ -75,6 +117,14 @@ public class MqttMessageHandler implements SmartLifecycle {
             qosArray[i] = 1;
         }
 
+        // 关键补充2：先取消原有订阅（避免重复订阅导致的消息重复接收）
+        try {
+            mqttClient.unsubscribe(topics);
+            log.info("【MQTT初始化】已取消原有主题订阅，准备重新订阅");
+        } catch (Exception e) {
+            log.warn("【MQTT初始化】取消原有订阅失败（首次订阅可忽略）：{}", e.getMessage());
+        }
+
         // 设置MQTT消息回调：处理连接断开、消息接收、消息发布完成
         mqttClient.setCallback(new MqttCallback() {
             /**
@@ -84,7 +134,11 @@ public class MqttMessageHandler implements SmartLifecycle {
             @Override
             public void connectionLost(Throwable cause) {
                 // 优化：替换System.err为log.error
-                log.error("【MQTT连接异常】连接断开：{}", cause.getMessage(), cause);
+                log.error("【MQTT连接异常】连接断开，clientId：{}，原因：{}", mqttClient.getClientId(), cause.getMessage(), cause);
+                // 新增：触发自动重连（仅当客户端处于运行状态且非配置错误时）
+                if (isRunning.get() && !isConfigError) {
+                    startReconnect();
+                }
             }
 
             /**
@@ -107,15 +161,18 @@ public class MqttMessageHandler implements SmartLifecycle {
             public void deliveryComplete(IMqttDeliveryToken token) {
                 // 优化：替换System.out为log.info，增加空值校验
                 if (token != null && token.getTopics() != null && token.getTopics().length > 0) {
-                    log.info("【MQTT确认】消息发布完成，主题：{}", token.getTopics()[0]);
+                    log.info("【MQTT确认】消息发布完成，clientId：{}，主题：{}", mqttClient.getClientId(), token.getTopics()[0]);
                 }
             }
         });
 
         // 订阅主题
         mqttClient.subscribe(topics, qosArray);
-        // 优化：替换System.out为log.info
-        log.info("【MQTT初始化】订阅主题：{}", String.join(",", topics));
+        // 重置重连计数和配置错误标记（连接成功后清零）
+        currentReconnectCount = 0;
+        isConfigError = false;
+        // 优化：打印clientId，方便排查重连后的实例是否替换成功
+        log.info("【MQTT初始化】订阅主题完成，clientId：{}，订阅主题：{}", mqttClient.getClientId(), String.join(",", topics));
     }
 
     /**
@@ -367,6 +424,148 @@ public class MqttMessageHandler implements SmartLifecycle {
         return keys;
     }
 
+    // ========== 新增：自动重连核心方法 ==========
+    /**
+     * 启动自动重连逻辑
+     */
+    private void startReconnect() {
+        // 终止条件：配置错误 或 达到最大重连次数
+        if (isConfigError || (maxReconnectTimes > 0 && currentReconnectCount >= maxReconnectTimes)) {
+            String reason = isConfigError ? "配置错误" : String.format("达到最大重连次数(%d)", maxReconnectTimes);
+            log.error("【MQTT重连】{}，停止重连", reason);
+            // isRunning.set(false);
+            return;
+        }
+        // 极端场景兜底：客户端实例为空则终止重连
+        if (mqttClient == null) {
+            log.error("【MQTT重连】客户端实例为空，终止重连");
+            return;
+        }
+
+        // 提交重连任务到单线程池
+        reconnectExecutor.schedule(() -> {
+            try {
+                currentReconnectCount++;
+                log.info("【MQTT重连】第{}次尝试重连（间隔{}秒）", currentReconnectCount, reconnectInterval);
+
+                // 兼容极端场景：先强制断开（无论当前状态），再重新连接
+                try {
+                    if (mqttClient.isConnected()) {
+                        mqttClient.disconnect();
+                        log.info("【MQTT重连】已断开旧连接");
+                    }
+                } catch (Exception e) {
+                    log.warn("【MQTT重连】断开旧连接失败（忽略）：{}", e.getMessage());
+                }
+
+                // 2. 生成新的clientId（和MqttConfig中一致的规则：原clientId + 随机后缀）
+                // 新代码（字母数字混合，和配置文件对齐）
+                String oldClientId = mqttClient.getClientId();
+                String originalClientId = oldClientId.contains("_") ? oldClientId.split("_")[0] : oldClientId;
+                // 调用工具方法生成8位字母数字随机串
+                String shortRandom = UUID.generateAlphanumericRandom(8);
+                String newClientId = originalClientId + "_" + shortRandom;
+
+                // 3. 创建新的MqttClient实例
+                MqttClient newMqttClient = new MqttClient(
+                        mqttClient.getServerURI(),
+                        newClientId,
+                        new MemoryPersistence()
+                );
+
+                // 4. 用新客户端连接
+                newMqttClient.connect(mqttConnectOptions);
+                log.info("【MQTT重连】使用新clientId连接成功：{}", newClientId);
+
+                // 5. 替换旧客户端实例，重新订阅
+                this.mqttClient = newMqttClient;
+                // 重新连接MQTT Broker + 重新订阅主题
+                subscribeTopics();
+
+                log.info("【MQTT重连】第{}次重连成功，新clientId：{}", currentReconnectCount, newClientId);
+
+            } catch (MqttException e) {
+                log.error("【MQTT重连】第{}次重连失败：{}", currentReconnectCount, e.getMessage(), e);
+                // 判断是否为配置类错误（永久错误）
+                judgeConfigError(e);
+                // 非配置错误则继续重连
+                if (!isConfigError) {
+                    startReconnect();
+                }
+            }
+        }, reconnectInterval, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 判断MQTT错误类型（配置错误/网络错误）
+     * 用纯数字错误码，兼容所有Paho版本
+     */
+    private void judgeConfigError(MqttException e) {
+        int errorCode = e.getReasonCode();
+        switch (errorCode) {
+            // 21 = 认证失败（用户名/密码错误）→ 配置错误
+            case 21:
+                isConfigError = true;
+                log.error("【MQTT配置错误】认证失败！请检查用户名/密码，错误码：{}", errorCode);
+                break;
+            // 32 = 客户端ID非法 → 配置错误
+            case 32:
+                isConfigError = true;
+                log.error("【MQTT配置错误】客户端ID非法！请检查clientId配置，错误码：{}", errorCode);
+                break;
+            // 3 = 连接被拒绝（地址/端口错误）→ 配置错误
+            case 3:
+                isConfigError = true;
+                log.error("【MQTT配置错误】连接被拒绝！请检查Broker地址/端口，错误码：{}", errorCode);
+                break;
+            // 31 = Broker不可达（网络波动）→ 继续重连
+            case 31:
+                log.warn("【MQTT错误】Broker不可达（网络波动），错误码：{}，继续重连", errorCode);
+                break;
+            // 其他错误 → 默认视为临时错误
+            default:
+                log.warn("【MQTT未知错误】错误码：{}，继续重连", errorCode);
+                break;
+        }
+    }
+
+    // ========== 新增：手动重连接口（供Controller调用） ==========
+    /**
+     * 手动触发MQTT重连（重置配置错误标记，强制重连）
+     */
+    public String manualReconnect() {
+        // 重置配置错误标记（允许重连）
+        isConfigError = false;
+        currentReconnectCount = 0;
+        isRunning.set(true);
+
+        try {
+            // 强制断开旧连接（如果存在）
+            if (mqttClient.isConnected()) {
+                mqttClient.disconnect();
+            }
+            // 重新初始化订阅
+            subscribeTopics();
+            log.info("【手动重连】MQTT客户端重连成功");
+            return "MQTT手动重连成功";
+        } catch (MqttException e) {
+            log.error("【手动重连】MQTT客户端重连失败", e);
+            judgeConfigError(e);
+            return "MQTT手动重连失败：" + e.getMessage();
+        }
+    }
+
+    /**
+     * 获取当前MQTT连接状态
+     */
+    public String getMqttStatus() {
+        boolean connected = mqttClient.isConnected();
+        String status = connected ? "已连接" : "已断开";
+        String configErrorStatus = isConfigError ? "（配置错误，已终止自动重连）" : "";
+        String reconnectStatus = String.format("当前重连次数：%d，最大重连次数：%d", currentReconnectCount, maxReconnectTimes);
+        return String.format("MQTT连接状态：%s%s；%s", status, configErrorStatus, reconnectStatus);
+    }
+
     // ======================== SmartLifecycle 生命周期管理（核心修复） ========================
     /**
      * 启动MQTT客户端（Spring上下文初始化/重启时触发）
@@ -374,14 +573,22 @@ public class MqttMessageHandler implements SmartLifecycle {
      */
     @Override
     public void start() {
+        log.info("开始监听");
         if (isRunning.compareAndSet(false, true)) {
+            log.info("开始监听111");
             try {
-                // 重新初始化MQTT订阅和回调
+                // 核心修改：无论是否已连接，都执行订阅（设置回调+订阅主题）
+                // 移除原有的if (!mqttClient.isConnected()) 判断
+                log.info("开始监听222");
                 subscribeTopics();
-                log.info("【MQTT生命周期】客户端启动成功");
+                log.info("【MQTT生命周期】客户端启动成功（已设置回调+订阅主题）");
             } catch (MqttException e) {
                 log.error("【MQTT生命周期】客户端启动失败", e);
                 isRunning.set(false);
+                judgeConfigError(e); // 启动失败时判断配置错误
+                if (!isConfigError) {
+                    startReconnect();
+                }
             }
         }
     }
@@ -395,6 +602,12 @@ public class MqttMessageHandler implements SmartLifecycle {
         // 修复：JDK 8正确的compareAndSet写法（无命名参数）
         if (isRunning.compareAndSet(true, false)) {
             try {
+                // 新增：关闭重连线程池，避免内存泄漏
+                reconnectExecutor.shutdown();
+                if (!reconnectExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+                    reconnectExecutor.shutdownNow();
+                }
+
                 if (mqttClient != null && mqttClient.isConnected()) {
                     // 移除：取消订阅相关逻辑（避免依赖不存在的方法）
                     // 直接断开连接（基础Paho支持的核心方法）
@@ -405,7 +618,10 @@ public class MqttMessageHandler implements SmartLifecycle {
                 }
                 // 移除：resetConnection()（避免Redis版本差异）
                 // 替代方案：无需主动重置，Spring上下文重启会重新创建Redis连接
-            } catch (MqttException e) {
+                // 新增：重置重连相关状态
+                currentReconnectCount = 0;
+                isConfigError = false;
+            } catch (MqttException | InterruptedException e) {
                 log.error("【MQTT生命周期】客户端关闭失败", e);
             }
         }