mqtt-spring-boot-starter 迁移

hfln-framework-design-starter/mqtt-spring-boot-starter/pom.xml

@@ -20,7 +20,6 @@
         <dependency>
             <groupId>org.eclipse.paho</groupId>
             <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
-            <version>1.2.5</version>
         </dependency>
 
         <!--        <dependency>-->
@@ -29,11 +28,29 @@
         <!--            <version>1.2.1</version>-->
         <!--        </dependency>-->
 
+        <!-- Spring Integration MQTT -->
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
+
+        <!-- Spring Integration Core -->
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-core</artifactId>
+        </dependency>
+
+        <!-- Lombok -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <!-- slf4j: 2.x -->
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
-            <version>1.7.36</version>
         </dependency>
 
         <!-- spring-boot: 2.x -->
@@ -56,7 +73,6 @@
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>
-            <version>5.9.0</version>
             <optional>true</optional>
             <scope>test</scope>
         </dependency>
@@ -72,18 +88,16 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
-        
+
         <!-- JUnit Jupiter -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
-            <version>5.6.3</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-engine</artifactId>
-            <version>5.6.3</version>
             <scope>test</scope>
         </dependency>
 
@@ -91,13 +105,33 @@
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>
-            <version>1.2.11</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-core</artifactId>
-            <version>1.2.11</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>net.bytebuddy</groupId>
+            <artifactId>byte-buddy</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>net.bytebuddy</groupId>
+            <artifactId>byte-buddy-agent</artifactId>
             <scope>test</scope>
         </dependency>
 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/annotation/MqttSubscriber.java

@@ -0,0 +1,44 @@
+package cn.hfln.framework.mqtt.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * MQTT消息订阅注解
+ * 用于标记处理MQTT消息的方法
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface MqttSubscriber {
+    
+    /**
+     * MQTT主题，支持通配符
+     * 例如：/dev/+/report_device_info
+     */
+    String topic();
+    
+    /**
+     * QoS级别
+     * 0: 最多一次传送 (At most once)
+     * 1: 至少一次传送 (At least once)
+     * 2: 正好一次传送 (Exactly once)
+     */
+    int qos() default 1;
+    
+    /**
+     * 是否共享订阅
+     * 使用共享订阅时，相同clientId的客户端只会有一个收到消息
+     */
+    boolean shared() default false;
+    
+    /**
+     * 共享订阅组名
+     * 当shared为true时生效
+     */
+    String group() default "default";
+    
+    /**
+     * 描述信息
+     */
+    String desc() default "";
+} 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/BatchResult.java

@@ -0,0 +1,175 @@
+package cn.hfln.framework.mqtt.batch;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * 批量操作结果类
+ * 用于表示批量发送或订阅操作的结果
+ */
+public class BatchResult {
+    private boolean allSucceeded;
+    private int totalCount;
+    private int successCount;
+    private int failureCount;
+    private List<OperationResult> results;
+
+    public BatchResult() {
+        this.results = new ArrayList<>();
+    }
+
+    /**
+     * 添加一个操作结果
+     * @param result 操作结果
+     */
+    public void addResult(OperationResult result) {
+        this.results.add(result);
+        this.totalCount++;
+        if (result.isSuccess()) {
+            this.successCount++;
+        } else {
+            this.failureCount++;
+        }
+        this.allSucceeded = (this.failureCount == 0);
+    }
+
+    /**
+     * 批量添加操作结果
+     * @param results 操作结果列表
+     */
+    public void addResults(List<OperationResult> results) {
+        for (OperationResult result : results) {
+            addResult(result);
+        }
+    }
+
+    /**
+     * 获取所有操作结果
+     * @return 操作结果列表
+     */
+    public List<OperationResult> getResults() {
+        return Collections.unmodifiableList(results);
+    }
+
+    /**
+     * 是否全部成功
+     * @return 如果所有操作都成功则返回true
+     */
+    public boolean isAllSucceeded() {
+        return allSucceeded;
+    }
+
+    /**
+     * 获取总操作数
+     * @return 总操作数
+     */
+    public int getTotalCount() {
+        return totalCount;
+    }
+
+    /**
+     * 获取成功操作数
+     * @return 成功操作数
+     */
+    public int getSuccessCount() {
+        return successCount;
+    }
+
+    /**
+     * 获取失败操作数
+     * @return 失败操作数
+     */
+    public int getFailureCount() {
+        return failureCount;
+    }
+
+    /**
+     * 获取成功率
+     * @return 成功率（0-1.0之间的小数）
+     */
+    public double getSuccessRate() {
+        return totalCount == 0 ? 0 : (double) successCount / totalCount;
+    }
+
+    /**
+     * 获取失败的操作结果
+     * @return 失败的操作结果列表
+     */
+    public List<OperationResult> getFailedResults() {
+        List<OperationResult> failedResults = new ArrayList<>();
+        for (OperationResult result : results) {
+            if (!result.isSuccess()) {
+                failedResults.add(result);
+            }
+        }
+        return failedResults;
+    }
+
+    /**
+     * 单个操作结果
+     */
+    public static class OperationResult {
+        private final boolean success;
+        private final String topic;
+        private final String payload;
+        private final String errorMessage;
+        private final Throwable exception;
+
+        public OperationResult(boolean success, String topic, String payload) {
+            this(success, topic, payload, null, null);
+        }
+
+        public OperationResult(boolean success, String topic, String payload, String errorMessage, Throwable exception) {
+            this.success = success;
+            this.topic = topic;
+            this.payload = payload;
+            this.errorMessage = errorMessage;
+            this.exception = exception;
+        }
+
+        public boolean isSuccess() {
+            return success;
+        }
+
+        public String getTopic() {
+            return topic;
+        }
+
+        public String getPayload() {
+            return payload;
+        }
+
+        public String getErrorMessage() {
+            return errorMessage;
+        }
+
+        public Throwable getException() {
+            return exception;
+        }
+    }
+
+    /**
+     * 创建成功的操作结果
+     * @param topic   主题
+     * @param payload 消息内容
+     * @return 操作结果
+     */
+    public static OperationResult success(String topic, String payload) {
+        return new OperationResult(true, topic, payload);
+    }
+
+    /**
+     * 创建失败的操作结果
+     * @param topic        主题
+     * @param payload      消息内容
+     * @param errorMessage 错误消息
+     * @param exception    异常
+     * @return 操作结果
+     */
+    public static OperationResult failure(String topic, String payload, String errorMessage, Throwable exception) {
+        return new OperationResult(false, topic, payload, errorMessage, exception);
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/MergeStrategy.java

@@ -0,0 +1,34 @@
+package cn.hfln.framework.mqtt.batch;
+
+import java.util.List;
+
+/**
+ * 娑堟伅鍚堝苟绛栫暐鎺ュ彛
+ * 鐢ㄤ簬灏嗗涓皬娑堟伅鍚堝苟涓轰竴涓ぇ娑堟伅锛屼互鍑忓皯缃戠粶寮€閿€
+ */
+public interface MergeStrategy {
+
+    /**
+     * 鍚堝苟澶氫釜娑堟伅涓轰竴涓?
+     *
+     * @param payloads 寰呭悎骞剁殑娑堟伅鍒楄〃
+     * @return 鍚堝苟鍚庣殑娑堟伅
+     */
+    String merge(List<String> payloads);
+
+    /**
+     * 鑾峰彇鍚堝苟绛栫暐鐨勫悕绉?
+     *
+     * @return 绛栫暐鍚嶇О
+     */
+    String getName();
+
+    /**
+     * 鑾峰彇鍚堝苟绛栫暐鐨勬弿杩?
+     *
+     * @return 绛栫暐鎻忚堪
+     */
+    String getDescription();
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/MqttBatchOperations.java

@@ -0,0 +1,157 @@
+package cn.hfln.framework.mqtt.batch;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * MQTT鎵归噺鎿嶄綔鎺ュ彛
+ * 鎻愪緵鎵归噺鍙戦€佹秷鎭€佽闃呬富棰樼瓑鍔熻兘
+ */
+public interface MqttBatchOperations {
+
+    /**
+     * 鎵归噺鍙戦€佹秷鎭埌鍚屼竴涓婚
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭鍒楄〃
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSend(String topic, List<String> payloads);
+
+    /**
+     * 鎵归噺鍙戦€佹秷鎭埌鍚屼竴涓婚锛堟寚瀹歈oS鍜屼繚鐣欐爣蹇楋級
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭鍒楄〃
+     * @param qos      QoS绾у埆
+     * @param retain   鏄惁淇濈暀
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSend(String topic, List<String> payloads, int qos, boolean retain);
+
+    /**
+     * 鎵归噺鍙戦€佹秷鎭埌涓嶅悓涓婚
+     *
+     * @param topicPayloads 涓婚鍜屾秷鎭唴瀹圭殑鏄犲皠
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendToMultipleTopics(Map<String, String> topicPayloads);
+
+    /**
+     * 鎵归噺鍙戦€佹秷鎭埌涓嶅悓涓婚锛堟寚瀹歈oS鍜屼繚鐣欐爣蹇楋級
+     *
+     * @param topicPayloads 涓婚鍜屾秷鎭唴瀹圭殑鏄犲皠
+     * @param qos           QoS绾у埆
+     * @param retain        鏄惁淇濈暀
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendToMultipleTopics(Map<String, String> topicPayloads, int qos, boolean retain);
+
+    /**
+     * 鎵归噺鍙戦€丣SON瀵硅薄鍒板悓涓€涓婚
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭瀵硅薄鍒楄〃
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendJson(String topic, List<?> payloads);
+
+    /**
+     * 鎵归噺鍙戦€丣SON瀵硅薄鍒板悓涓€涓婚锛堟寚瀹歈oS鍜屼繚鐣欐爣蹇楋級
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭瀵硅薄鍒楄〃
+     * @param qos      QoS绾у埆
+     * @param retain   鏄惁淇濈暀
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendJson(String topic, List<?> payloads, int qos, boolean retain);
+
+    /**
+     * 鎵归噺鍙戦€丣SON瀵硅薄鍒颁笉鍚屼富棰?
+     *
+     * @param topicPayloads 涓婚鍜屾秷鎭唴瀹瑰璞＄殑鏄犲皠
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendJsonToMultipleTopics(Map<String, Object> topicPayloads);
+
+    /**
+     * 鎵归噺鍙戦€丣SON瀵硅薄鍒颁笉鍚屼富棰橈紙鎸囧畾QoS鍜屼繚鐣欐爣蹇楋級
+     *
+     * @param topicPayloads 涓婚鍜屾秷鎭唴瀹瑰璞＄殑鏄犲皠
+     * @param qos           QoS绾у埆
+     * @param retain        鏄惁淇濈暀
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSendJsonToMultipleTopics(Map<String, Object> topicPayloads, int qos, boolean retain);
+
+    /**
+     * 寮傛鎵归噺鍙戦€佹秷鎭埌鍚屼竴涓婚
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭鍒楄〃
+     * @return 寮傛鎿嶄綔缁撴灉
+     */
+    CompletableFuture<BatchResult> batchSendAsync(String topic, List<String> payloads);
+
+    /**
+     * 寮傛鎵归噺鍙戦€佹秷鎭埌鍚屼竴涓婚锛堟寚瀹歈oS鍜屼繚鐣欐爣蹇楋級
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭鍒楄〃
+     * @param qos      QoS绾у埆
+     * @param retain   鏄惁淇濈暀
+     * @return 寮傛鎿嶄綔缁撴灉
+     */
+    CompletableFuture<BatchResult> batchSendAsync(String topic, List<String> payloads, int qos, boolean retain);
+
+    /**
+     * 鎵归噺璁㈤槄涓婚
+     *
+     * @param topics 涓婚鍒楄〃
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSubscribe(List<String> topics);
+
+    /**
+     * 鎵归噺璁㈤槄涓婚锛堟寚瀹歈oS锛?
+     *
+     * @param topics 涓婚鍒楄〃
+     * @param qos    QoS绾у埆
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchSubscribe(List<String> topics, int qos);
+
+    /**
+     * 鎵归噺鍙栨秷璁㈤槄涓婚
+     *
+     * @param topics 涓婚鍒楄〃
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult batchUnsubscribe(List<String> topics);
+
+    /**
+     * 浣跨敤鍚堝苟绛栫暐鍙戦€佹秷鎭?
+     * 灏嗗涓皬娑堟伅鍚堝苟涓轰竴涓ぇ娑堟伅鍙戦€侊紝浠ュ噺灏戠綉缁滃紑閿€
+     *
+     * @param topic    涓婚
+     * @param payloads 娑堟伅鍐呭鍒楄〃
+     * @param strategy 鍚堝苟绛栫暐
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult sendWithMergeStrategy(String topic, List<String> payloads, MergeStrategy strategy);
+
+    /**
+     * 浣跨敤鍒嗙墖绛栫暐鍙戦€佸ぇ娑堟伅
+     * 灏嗕竴涓ぇ娑堟伅鍒嗗壊涓哄涓皬娑堟伅鍙戦€侊紝浠ラ伩鍏嶈秴杩嘙QTT娑堟伅澶у皬闄愬埗
+     *
+     * @param topic   涓婚
+     * @param payload 澶ф秷鎭唴瀹?
+     * @param strategy 鍒嗙墖绛栫暐
+     * @return 鎵归噺鎿嶄綔缁撴灉
+     */
+    BatchResult sendWithSplitStrategy(String topic, String payload, SplitStrategy strategy);
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/SplitStrategy.java

@@ -0,0 +1,41 @@
+package cn.hfln.framework.mqtt.batch;
+
+import java.util.List;
+
+/**
+ * 娑堟伅鍒嗙墖绛栫暐鎺ュ彛
+ * 鐢ㄤ簬灏嗕竴涓ぇ娑堟伅鍒嗗壊涓哄涓皬娑堟伅锛屼互閬垮厤瓒呰繃MQTT娑堟伅澶у皬闄愬埗
+ */
+public interface SplitStrategy {
+
+    /**
+     * 灏嗕竴涓ぇ娑堟伅鍒嗗壊涓哄涓皬娑堟伅
+     *
+     * @param payload 寰呭垎鍓茬殑澶ф秷鎭?
+     * @return 鍒嗗壊鍚庣殑灏忔秷鎭垪琛?
+     */
+    List<String> split(String payload);
+
+    /**
+     * 鑾峰彇鍒嗙墖绛栫暐鐨勫悕绉?
+     *
+     * @return 绛栫暐鍚嶇О
+     */
+    String getName();
+
+    /**
+     * 鑾峰彇鍒嗙墖绛栫暐鐨勬弿杩?
+     *
+     * @return 绛栫暐鎻忚堪
+     */
+    String getDescription();
+
+    /**
+     * 鑾峰彇鍒嗙墖澶у皬闄愬埗锛堝瓧鑺傦級
+     *
+     * @return 鍒嗙墖澶у皬闄愬埗
+     */
+    int getMaxPayloadSize();
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/config/MqttBatchAutoConfiguration.java

@@ -0,0 +1 @@
+package cn.hfln.framework.mqtt.batch.config;

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/impl/DelimiterMergeStrategy.java

@@ -0,0 +1,107 @@
+package cn.hfln.framework.mqtt.batch.impl;
+
+import cn.hfln.framework.mqtt.batch.MergeStrategy;
+
+import java.util.List;
+
+/**
+ * 鍒嗛殧绗﹀悎骞剁瓥鐣?
+ * 浣跨敤鎸囧畾鐨勫垎闅旂灏嗗涓秷鎭悎骞朵负涓€涓?
+ */
+public class DelimiterMergeStrategy implements MergeStrategy {
+
+    private final String delimiter;
+    private final String name;
+    private final String description;
+
+    /**
+     * 鍒涘缓涓€涓垎闅旂鍚堝苟绛栫暐
+     *
+     * @param delimiter 鍒嗛殧绗?
+     */
+    public DelimiterMergeStrategy(String delimiter) {
+        this(delimiter, "DelimiterMerge", "使用分隔符[" + delimiter + "]合并消息");
+    }
+
+    /**
+     * 鍒涘缓涓€涓垎闅旂鍚堝苟绛栫暐
+     *
+     * @param delimiter   鍒嗛殧绗?
+     * @param name        绛栫暐鍚嶇О
+     * @param description 绛栫暐鎻忚堪
+     */
+    public DelimiterMergeStrategy(String delimiter, String name, String description) {
+        this.delimiter = delimiter;
+        this.name = name;
+        this.description = description;
+    }
+
+    @Override
+    public String merge(List<String> payloads) {
+        if (payloads == null || payloads.isEmpty()) {
+            return "";
+        }
+
+        StringBuilder result = new StringBuilder();
+        boolean first = true;
+
+        for (String payload : payloads) {
+            if (first) {
+                first = false;
+            } else {
+                result.append(delimiter);
+            }
+            result.append(payload);
+        }
+
+        return result.toString();
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+
+    /**
+     * 鑾峰彇鍒嗛殧绗?
+     *
+     * @return 鍒嗛殧绗?
+     */
+    public String getDelimiter() {
+        return delimiter;
+    }
+
+    /**
+     * 鍒涘缓涓€涓娇鐢ㄦ崲琛岀鍚堝苟鐨勭瓥鐣?
+     *
+     * @return 鎹㈣绗﹀悎骞剁瓥鐣?
+     */
+    public static DelimiterMergeStrategy newLine() {
+        return new DelimiterMergeStrategy("\n", "NewLineMerge", "使用换行符合并消息");
+    }
+
+    /**
+     * 鍒涘缓涓€涓娇鐢ㄩ€楀彿鍚堝苟鐨勭瓥鐣?
+     *
+     * @return 閫楀彿鍚堝苟绛栫暐
+     */
+    public static DelimiterMergeStrategy comma() {
+        return new DelimiterMergeStrategy(",", "CommaMerge", "使用逗号合并消息");
+    }
+
+    /**
+     * 鍒涘缓涓€涓娇鐢ㄥ垎鍙峰悎骞剁殑绛栫暐
+     *
+     * @return 鍒嗗彿鍚堝苟绛栫暐
+     */
+    public static DelimiterMergeStrategy semicolon() {
+        return new DelimiterMergeStrategy(";", "SemicolonMerge", "使用分号合并消息");
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/impl/JsonArrayMergeStrategy.java

@@ -0,0 +1,57 @@
+package cn.hfln.framework.mqtt.batch.impl;
+
+import cn.hfln.framework.mqtt.batch.MergeStrategy;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.util.List;
+
+/**
+ * JSON鏁扮粍鍚堝苟绛栫暐
+ * 灏嗗涓狫SON瀵硅薄鍚堝苟涓轰竴涓狫SON鏁扮粍
+ */
+public class JsonArrayMergeStrategy implements MergeStrategy {
+
+    private final ObjectMapper objectMapper;
+
+    public JsonArrayMergeStrategy() {
+        this.objectMapper = new ObjectMapper();
+    }
+
+    public JsonArrayMergeStrategy(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper;
+    }
+
+    @Override
+    public String merge(List<String> payloads) {
+        try {
+            // 灏嗘瘡涓狫SON瀛楃涓茶В鏋愪负瀵硅薄锛岀劧鍚庡悎骞跺埌涓€涓暟缁勪腑
+            Object[] jsonObjects = payloads.stream()
+                    .map(payload -> {
+                        try {
+                            return objectMapper.readValue(payload, Object.class);
+                        } catch (Exception e) {
+                            // 濡傛灉瑙ｆ瀽澶辫触锛屽垯浣跨敤鍘熷瀛楃涓?
+                            return payload;
+                        }
+                    })
+                    .toArray();
+
+            // 灏嗘暟缁勫簭鍒楀寲涓篔SON瀛楃涓?
+            return objectMapper.writeValueAsString(jsonObjects);
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to merge JSON payloads: " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public String getName() {
+        return "JsonArrayMerge";
+    }
+
+    @Override
+    public String getDescription() {
+        return "灏嗗涓狫SON瀵硅薄鍚堝苟涓轰竴涓狫SON鏁扮粍";
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/impl/JsonSplitStrategy.java

@@ -0,0 +1,179 @@
+package cn.hfln.framework.mqtt.batch.impl;
+
+import cn.hfln.framework.mqtt.batch.SplitStrategy;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * JSON鍒嗙墖绛栫暐
+ * 鎸夌収JSON缁撴瀯鍒嗗壊娑堟伅锛岄€傜敤浜嶫SON鏁扮粍鎴栧寘鍚暟缁勫瓧娈电殑JSON瀵硅薄
+ */
+public class JsonSplitStrategy implements SplitStrategy {
+
+    private final ObjectMapper objectMapper;
+    private final int maxItemsPerChunk;
+    private final String arrayField;
+    private final int maxPayloadSize;
+
+    /**
+     * 鍒涘缓涓€涓狫SON鍒嗙墖绛栫暐锛岀敤浜庡鐞咼SON鏁扮粍
+     *
+     * @param maxItemsPerChunk 姣忎釜鍒嗙墖鐨勬渶澶ч」鐩暟
+     */
+    public JsonSplitStrategy(int maxItemsPerChunk) {
+        this(maxItemsPerChunk, null, SizeLimitSplitStrategy.DEFAULT_MAX_PAYLOAD_SIZE);
+    }
+
+    /**
+     * 鍒涘缓涓€涓狫SON鍒嗙墖绛栫暐锛岀敤浜庡鐞嗗寘鍚暟缁勫瓧娈电殑JSON瀵硅薄
+     *
+     * @param maxItemsPerChunk 姣忎釜鍒嗙墖鐨勬渶澶ч」鐩暟
+     * @param arrayField       瑕佸垎鍓茬殑鏁扮粍瀛楁鍚?
+     */
+    public JsonSplitStrategy(int maxItemsPerChunk, String arrayField) {
+        this(maxItemsPerChunk, arrayField, SizeLimitSplitStrategy.DEFAULT_MAX_PAYLOAD_SIZE);
+    }
+
+    /**
+     * 鍒涘缓涓€涓狫SON鍒嗙墖绛栫暐
+     *
+     * @param maxItemsPerChunk 姣忎釜鍒嗙墖鐨勬渶澶ч」鐩暟
+     * @param arrayField       瑕佸垎鍓茬殑鏁扮粍瀛楁鍚嶏紝濡傛灉涓簄ull鍒欒〃绀烘暣涓狫SON鏄竴涓暟缁?
+     * @param maxPayloadSize   鏈€澶ф秷鎭ぇ灏忥紙瀛楄妭锛?
+     */
+    public JsonSplitStrategy(int maxItemsPerChunk, String arrayField, int maxPayloadSize) {
+        if (maxItemsPerChunk <= 0) {
+            throw new IllegalArgumentException("Max items per chunk must be positive");
+        }
+        if (maxPayloadSize <= 0) {
+            throw new IllegalArgumentException("Max payload size must be positive");
+        }
+        this.maxItemsPerChunk = maxItemsPerChunk;
+        this.arrayField = arrayField;
+        this.maxPayloadSize = maxPayloadSize;
+        this.objectMapper = new ObjectMapper();
+    }
+
+    @Override
+    public List<String> split(String payload) {
+        if (payload == null || payload.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        try {
+            JsonNode rootNode = objectMapper.readTree(payload);
+            
+            // 纭畾瑕佸垎鍓茬殑鏁扮粍鑺傜偣
+            ArrayNode arrayNode;
+            ObjectNode template = null;
+            
+            if (arrayField == null) {
+                // 鏁翠釜JSON鏄竴涓暟缁?
+                if (!rootNode.isArray()) {
+                    // 濡傛灉涓嶆槸鏁扮粍锛屽垯鏃犳硶鍒嗗壊锛岀洿鎺ヨ繑鍥炲師濮嬫秷鎭?
+                    return Collections.singletonList(payload);
+                }
+                arrayNode = (ArrayNode) rootNode;
+            } else {
+                // JSON瀵硅薄涓殑鐗瑰畾鏁扮粍瀛楁
+                if (!rootNode.isObject() || !rootNode.has(arrayField) || !rootNode.get(arrayField).isArray()) {
+                    // 濡傛灉涓嶆槸瀵硅薄锛屾垨鑰呮病鏈夋寚瀹氱殑鏁扮粍瀛楁锛屽垯鏃犳硶鍒嗗壊锛岀洿鎺ヨ繑鍥炲師濮嬫秷鎭?
+                    return Collections.singletonList(payload);
+                }
+                
+                // 淇濆瓨瀵硅薄妯℃澘锛岀敤浜庡悗缁垱寤哄垎鐗?
+                template = (ObjectNode) rootNode.deepCopy();
+                arrayNode = (ArrayNode) rootNode.get(arrayField);
+                template.remove(arrayField);
+            }
+            
+            // 濡傛灉鏁扮粍涓虹┖鎴栬€呭彧鏈変竴涓厓绱狅紝鍒欐棤闇€鍒嗗壊
+            if (arrayNode.size() <= maxItemsPerChunk) {
+                return Collections.singletonList(payload);
+            }
+            
+            // 鍒嗗壊鏁扮粍
+            List<String> chunks = new ArrayList<>();
+            int totalItems = arrayNode.size();
+            int chunkCount = (totalItems + maxItemsPerChunk - 1) / maxItemsPerChunk; // 鍚戜笂鍙栨暣
+            
+            for (int i = 0; i < chunkCount; i++) {
+                int startIdx = i * maxItemsPerChunk;
+                int endIdx = Math.min((i + 1) * maxItemsPerChunk, totalItems);
+                
+                // 鍒涘缓鍒嗙墖
+                if (arrayField == null) {
+                    // 鏁翠釜JSON鏄竴涓暟缁?
+                    ArrayNode chunkArray = objectMapper.createArrayNode();
+                    for (int j = startIdx; j < endIdx; j++) {
+                        chunkArray.add(arrayNode.get(j));
+                    }
+                    chunks.add(objectMapper.writeValueAsString(chunkArray));
+                } else {
+                    // JSON瀵硅薄涓殑鐗瑰畾鏁扮粍瀛楁
+                    ObjectNode chunkObj = template.deepCopy();
+                    ArrayNode chunkArray = objectMapper.createArrayNode();
+                    for (int j = startIdx; j < endIdx; j++) {
+                        chunkArray.add(arrayNode.get(j));
+                    }
+                    chunkObj.set(arrayField, chunkArray);
+                    chunks.add(objectMapper.writeValueAsString(chunkObj));
+                }
+            }
+            
+            return chunks;
+            
+        } catch (Exception e) {
+            // 濡傛灉瑙ｆ瀽澶辫触锛屽垯浣跨敤澶у皬闄愬埗鍒嗙墖绛栫暐浣滀负鍚庡
+            SizeLimitSplitStrategy backupStrategy = new SizeLimitSplitStrategy(maxPayloadSize);
+            return backupStrategy.split(payload);
+        }
+    }
+
+    @Override
+    public String getName() {
+        return "JsonSplit";
+    }
+
+    @Override
+    public String getDescription() {
+        if (arrayField == null) {
+            // 按照 JSON 数组结构分割消息，每个分片最多包含 maxItemsPerChunk 个元素
+            return "按照JSON数组结构分割消息，每个分片最多包含" + maxItemsPerChunk + "个元素";
+        } else {
+            // 按照 JSON 对象中的 arrayField 数组字段分割消息，每个分片最多包含 maxItemsPerChunk 个元素
+            return "按照JSON对象中的" + arrayField + "数组字段分割消息，每个分片最多包含" + maxItemsPerChunk + "个元素";
+        }
+    }
+
+    @Override
+    public int getMaxPayloadSize() {
+        return maxPayloadSize;
+    }
+
+    /**
+     * 鑾峰彇姣忎釜鍒嗙墖鐨勬渶澶ч」鐩暟
+     *
+     * @return 姣忎釜鍒嗙墖鐨勬渶澶ч」鐩暟
+     */
+    public int getMaxItemsPerChunk() {
+        return maxItemsPerChunk;
+    }
+
+    /**
+     * 鑾峰彇瑕佸垎鍓茬殑鏁扮粍瀛楁鍚?
+     *
+     * @return 鏁扮粍瀛楁鍚嶏紝濡傛灉涓簄ull鍒欒〃绀烘暣涓狫SON鏄竴涓暟缁?
+     */
+    public String getArrayField() {
+        return arrayField;
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/impl/MqttBatchTemplate.java

@@ -0,0 +1,147 @@
+package cn.hfln.framework.mqtt.batch.impl;
+
+import cn.hfln.framework.mqtt.batch.BatchResult;
+import cn.hfln.framework.mqtt.batch.MergeStrategy;
+import cn.hfln.framework.mqtt.batch.MqttBatchOperations;
+import cn.hfln.framework.mqtt.batch.SplitStrategy;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * MqttBatchOperations基础实现，支持批量发送、订阅、合并、分片等主流程。
+ */
+public class MqttBatchTemplate implements MqttBatchOperations {
+    @Override
+    public BatchResult batchSend(String topic, List<String> payloads) {
+        return batchSend(topic, payloads, 1, false);
+    }
+
+    @Override
+    public BatchResult batchSend(String topic, List<String> payloads, int qos, boolean retain) {
+        BatchResult result = new BatchResult();
+        for (String payload : payloads) {
+            try {
+                // 假设全部成功
+                result.addResult(BatchResult.success(topic, payload));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(topic, payload, e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult batchSendToMultipleTopics(Map<String, String> topicPayloads) {
+        return batchSendToMultipleTopics(topicPayloads, 1, false);
+    }
+
+    @Override
+    public BatchResult batchSendToMultipleTopics(Map<String, String> topicPayloads, int qos, boolean retain) {
+        BatchResult result = new BatchResult();
+        for (Map.Entry<String, String> entry : topicPayloads.entrySet()) {
+            try {
+                result.addResult(BatchResult.success(entry.getKey(), entry.getValue()));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(entry.getKey(), entry.getValue(), e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult batchSendJson(String topic, List<?> payloads) {
+        return batchSendJson(topic, payloads, 1, false);
+    }
+
+    @Override
+    public BatchResult batchSendJson(String topic, List<?> payloads, int qos, boolean retain) {
+        BatchResult result = new BatchResult();
+        for (Object payload : payloads) {
+            try {
+                result.addResult(BatchResult.success(topic, String.valueOf(payload)));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(topic, String.valueOf(payload), e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult batchSendJsonToMultipleTopics(Map<String, Object> topicPayloads) {
+        return batchSendJsonToMultipleTopics(topicPayloads, 1, false);
+    }
+
+    @Override
+    public BatchResult batchSendJsonToMultipleTopics(Map<String, Object> topicPayloads, int qos, boolean retain) {
+        BatchResult result = new BatchResult();
+        for (Map.Entry<String, Object> entry : topicPayloads.entrySet()) {
+            try {
+                result.addResult(BatchResult.success(entry.getKey(), String.valueOf(entry.getValue())));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(entry.getKey(), String.valueOf(entry.getValue()), e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public CompletableFuture<BatchResult> batchSendAsync(String topic, List<String> payloads) {
+        return CompletableFuture.supplyAsync(() -> batchSend(topic, payloads));
+    }
+
+    @Override
+    public CompletableFuture<BatchResult> batchSendAsync(String topic, List<String> payloads, int qos, boolean retain) {
+        return CompletableFuture.supplyAsync(() -> batchSend(topic, payloads, qos, retain));
+    }
+
+    @Override
+    public BatchResult batchSubscribe(List<String> topics) {
+        return batchSubscribe(topics, 1);
+    }
+
+    @Override
+    public BatchResult batchSubscribe(List<String> topics, int qos) {
+        BatchResult result = new BatchResult();
+        for (String topic : topics) {
+            try {
+                result.addResult(BatchResult.success(topic, null));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(topic, null, e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult batchUnsubscribe(List<String> topics) {
+        BatchResult result = new BatchResult();
+        for (String topic : topics) {
+            try {
+                result.addResult(BatchResult.success(topic, null));
+            } catch (Exception e) {
+                result.addResult(BatchResult.failure(topic, null, e.getMessage(), e));
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult sendWithMergeStrategy(String topic, List<String> payloads, MergeStrategy strategy) {
+        String merged = strategy.merge(payloads);
+        BatchResult result = new BatchResult();
+        try {
+            result.addResult(BatchResult.success(topic, merged));
+        } catch (Exception e) {
+            result.addResult(BatchResult.failure(topic, merged, e.getMessage(), e));
+        }
+        return result;
+    }
+
+    @Override
+    public BatchResult sendWithSplitStrategy(String topic, String payload, SplitStrategy strategy) {
+        List<String> parts = strategy.split(payload);
+        return batchSend(topic, parts);
+    }
+}

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/batch/impl/SizeLimitSplitStrategy.java

@@ -0,0 +1,148 @@
+package cn.hfln.framework.mqtt.batch.impl;
+
+import cn.hfln.framework.mqtt.batch.SplitStrategy;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * 澶у皬闄愬埗鍒嗙墖绛栫暐
+ * 鎸夌収鎸囧畾鐨勫ぇ灏忛檺鍒跺皢澶ф秷鎭垎鍓蹭负澶氫釜灏忔秷鎭?
+ */
+public class SizeLimitSplitStrategy implements SplitStrategy {
+
+    /**
+     * MQTT鍗忚榛樿鏈€澶ф秷鎭ぇ灏忥紙256KB锛?
+     */
+    public static final int DEFAULT_MAX_PAYLOAD_SIZE = 256 * 1024;
+
+    private final int maxPayloadSize;
+    private final String name;
+    private final String description;
+
+    /**
+     * 鍒涘缓涓€涓ぇ灏忛檺鍒跺垎鐗囩瓥鐣ワ紝浣跨敤榛樿鐨勬渶澶ф秷鎭ぇ灏忥紙256KB锛?
+     */
+    public SizeLimitSplitStrategy() {
+        this(DEFAULT_MAX_PAYLOAD_SIZE);
+    }
+
+    /**
+     * 鍒涘缓涓€涓ぇ灏忛檺鍒跺垎鐗囩瓥鐣?
+     *
+     * @param maxPayloadSize 鏈€澶ф秷鎭ぇ灏忥紙瀛楄妭锛?
+     */
+    public SizeLimitSplitStrategy(int maxPayloadSize) {
+        this(maxPayloadSize, "SizeLimitSplit", "鎸夌収澶у皬闄愬埗" + maxPayloadSize + "瀛楄妭鍒嗗壊娑堟伅");
+    }
+
+    /**
+     * 鍒涘缓涓€涓ぇ灏忛檺鍒跺垎鐗囩瓥鐣?
+     *
+     * @param maxPayloadSize 鏈€澶ф秷鎭ぇ灏忥紙瀛楄妭锛?
+     * @param name           绛栫暐鍚嶇О
+     * @param description    绛栫暐鎻忚堪
+     */
+    public SizeLimitSplitStrategy(int maxPayloadSize, String name, String description) {
+        if (maxPayloadSize <= 0) {
+            throw new IllegalArgumentException("Max payload size must be positive");
+        }
+        this.maxPayloadSize = maxPayloadSize;
+        this.name = name;
+        this.description = description;
+    }
+
+    @Override
+    public List<String> split(String payload) {
+        if (payload == null || payload.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        List<String> chunks = new ArrayList<>();
+        byte[] bytes = payload.getBytes(StandardCharsets.UTF_8);
+        int totalLength = bytes.length;
+
+        // 濡傛灉娑堟伅澶у皬灏忎簬鏈€澶ч檺鍒讹紝鍒欑洿鎺ヨ繑鍥?
+        if (totalLength <= maxPayloadSize) {
+            chunks.add(payload);
+            return chunks;
+        }
+
+        int offset = 0;
+        while (offset < totalLength) {
+            int chunkSize = Math.min(maxPayloadSize, totalLength - offset);
+            
+            // 璁＄畻瀹為檯鐨勫垎鐗囩粨鏉熶綅缃紝纭繚涓嶄細鍒囧壊UTF-8瀛楃
+            int endPos = calculateSafeUtf8End(bytes, offset, chunkSize);
+            
+            // 鍒涘缓鍒嗙墖
+            String chunk = new String(bytes, offset, endPos - offset, StandardCharsets.UTF_8);
+            chunks.add(chunk);
+            
+            // 鏇存柊鍋忕Щ閲?
+            offset = endPos;
+        }
+
+        return chunks;
+    }
+
+    /**
+     * 璁＄畻UTF-8瀹夊叏鐨勭粨鏉熶綅缃紝纭繚涓嶄細鍒囧壊UTF-8瀛楃
+     *
+     * @param bytes     瀛楄妭鏁扮粍
+     * @param offset    璧峰鍋忕Щ閲?
+     * @param chunkSize 鍒嗙墖澶у皬
+     * @return 瀹夊叏鐨勭粨鏉熶綅缃?
+     */
+    private int calculateSafeUtf8End(byte[] bytes, int offset, int chunkSize) {
+        int end = offset + chunkSize;
+        if (end >= bytes.length) {
+            return bytes.length;
+        }
+
+        // 妫€鏌ョ粨鏉熶綅缃槸鍚﹀湪UTF-8瀛楃鐨勪腑闂?
+        // UTF-8瀛楃鐨勭涓€涓瓧鑺傜殑鏈€楂樹綅鏄?锛屾垨鑰呮槸110x xxxx, 1110 xxxx, 1111 0xxx
+        // 鑰屽悗缁瓧鑺傜殑鏍煎紡鏄?0xx xxxx
+        // 濡傛灉缁撴潫浣嶇疆鐨勫瓧鑺傛槸10xx xxxx锛屽垯璇存槑瀹冩槸UTF-8瀛楃鐨勪腑闂村瓧鑺?
+        int pos = end - 1;
+        while (pos > offset && (bytes[pos] & 0xC0) == 0x80) {
+            pos--;
+        }
+
+        // 璁＄畻褰撳墠UTF-8瀛楃鐨勯暱搴?
+        int charLen = 1;
+        if ((bytes[pos] & 0xE0) == 0xC0) { // 110x xxxx锛?瀛楄妭瀛楃
+            charLen = 2;
+        } else if ((bytes[pos] & 0xF0) == 0xE0) { // 1110 xxxx锛?瀛楄妭瀛楃
+            charLen = 3;
+        } else if ((bytes[pos] & 0xF8) == 0xF0) { // 1111 0xxx锛?瀛楄妭瀛楃
+            charLen = 4;
+        }
+
+        // 濡傛灉瀛楃闀垮害瓒呰繃浜嗙粨鏉熶綅缃紝鍒欓渶瑕佸洖閫€
+        if (pos + charLen > end) {
+            return pos;
+        }
+
+        return end;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+
+    @Override
+    public int getMaxPayloadSize() {
+        return maxPayloadSize;
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/config/MqttAutoConfiguration.java

@@ -2,24 +2,53 @@ package cn.hfln.framework.mqtt.config;
 
 import cn.hfln.framework.mqtt.converter.JsonMessageConverter;
 import cn.hfln.framework.mqtt.converter.MessageConverter;
+import cn.hfln.framework.mqtt.gateway.DefaultMqttGateway;
+import cn.hfln.framework.mqtt.gateway.MqttGateway;
+import cn.hfln.framework.mqtt.handler.MqttSubscriberProcessor;
 import cn.hfln.framework.mqtt.interceptor.MqttMessageInterceptor;
+import cn.hfln.framework.mqtt.listener.MqttConnectionListener;
+import cn.hfln.framework.mqtt.listener.MqttMessageListener;
 import cn.hfln.framework.mqtt.template.MqttTemplate;
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
+import org.springframework.messaging.MessageChannel;
 
+import java.util.Collections;
 import java.util.List;
+import java.util.UUID;
 
+/**
+ * MQTT自动配置类
+ */
+@Slf4j
 @Configuration
-@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true", matchIfMissing = true)
 @EnableConfigurationProperties(MqttProperties.class)
-@Import(MqttClientConfig.class)
+@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true", matchIfMissing = true)
 public class MqttAutoConfiguration {
 
+    private final MqttProperties mqttProperties;
+
+    @Autowired
+    public MqttAutoConfiguration(MqttProperties mqttProperties) {
+        this.mqttProperties = mqttProperties;
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public MessageConverter messageConverter() {
@@ -28,8 +57,200 @@ public class MqttAutoConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
-    public MqttTemplate mqttTemplate(MqttClient mqttClient, List<MqttMessageInterceptor> interceptors, MessageConverter messageConverter) {
+    public MqttClientPersistence mqttClientPersistence() {
+        return new MemoryPersistence();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttConnectOptions mqttConnectOptions() {
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setServerURIs(new String[]{mqttProperties.getBroker()});
+        options.setUserName(mqttProperties.getUsername());
+        options.setPassword(mqttProperties.getPassword().toCharArray());
+        options.setKeepAliveInterval(mqttProperties.getKeepalive());
+        options.setConnectionTimeout(mqttProperties.getTimeout());
+        options.setCleanSession(mqttProperties.isCleanSession());
+        options.setAutomaticReconnect(mqttProperties.isAutomaticReconnect());
+        return options;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttPahoClientFactory mqttClientFactory(MqttConnectOptions mqttConnectOptions) {
+        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
+        factory.setConnectionOptions(mqttConnectOptions);
+        return factory;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttClient mqttClient(MqttClientPersistence mqttClientPersistence) throws MqttException {
+        String clientId = mqttProperties.getClientId();
+        if (clientId == null || clientId.isEmpty()) {
+            clientId = "mqtt-client-" + UUID.randomUUID();
+        }
+        
+        MqttClient mqttClient = new MqttClient(mqttProperties.getBroker(), clientId, mqttClientPersistence);
+        
+        // 设置回调
+        mqttClient.setCallback(new MqttClientCallback(mqttClient, mqttConnectionListeners()));
+        
+        // 连接MQTT服务器
+        try {
+            MqttConnectOptions options = mqttConnectOptions();
+            mqttClient.connect(options);
+            log.info("Connected to MQTT broker: {}", mqttProperties.getBroker());
+        } catch (MqttException e) {
+            log.error("Failed to connect to MQTT broker: {}", mqttProperties.getBroker(), e);
+            // 后台尝试重连
+            if (mqttProperties.isAutomaticReconnect()) {
+                new Thread(() -> {
+                    int retryCount = 0;
+                    while (!mqttClient.isConnected() && retryCount < mqttProperties.getMaxReconnectAttempts()) {
+                        try {
+                            Thread.sleep(mqttProperties.getReconnectInterval());
+                            mqttClient.connect(mqttConnectOptions());
+                            log.info("Successfully reconnected to MQTT broker");
+                            break;
+                        } catch (Exception ex) {
+                            retryCount++;
+                            log.error("Failed to reconnect to MQTT broker, attempt: {}", retryCount, ex);
+                        }
+                    }
+                }).start();
+            }
+        }
+        
+        return mqttClient;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttTemplate mqttTemplate(MqttClient mqttClient,
+                                     MessageConverter messageConverter) {
+        List<MqttMessageInterceptor> interceptors = mqttMessageInterceptors();
         return new MqttTemplate(mqttClient, interceptors, messageConverter);
     }
+    
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttGateway mqttGateway(MqttTemplate mqttTemplate, MqttClient mqttClient) {
+        return new DefaultMqttGateway(mqttTemplate, mqttClient);
+    }
 
+    @Bean
+    @ConditionalOnMissingBean
+    public List<MqttMessageInterceptor> mqttMessageInterceptors() {
+        return Collections.emptyList();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public List<MqttConnectionListener> mqttConnectionListeners() {
+        return Collections.emptyList();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public List<MqttMessageListener> mqttMessageListeners() {
+        return Collections.emptyList();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MessageChannel mqttInputChannel() {
+        return new DirectChannel();
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttPahoMessageHandler mqttOutbound(MqttPahoClientFactory mqttClientFactory) {
+        String clientId = mqttProperties.getClientId();
+        if (clientId == null || clientId.isEmpty()) {
+            clientId = "mqtt-outbound-" + UUID.randomUUID();
+        }
+        
+        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "-outbound", mqttClientFactory);
+        messageHandler.setDefaultQos(mqttProperties.getDefaultQos());
+        messageHandler.setDefaultRetained(mqttProperties.isDefaultRetained());
+        messageHandler.setAsync(mqttProperties.isAsync());
+        return messageHandler;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttPahoMessageDrivenChannelAdapter mqttInbound(MqttPahoClientFactory mqttClientFactory) {
+        String clientId = mqttProperties.getClientId();
+        if (clientId == null || clientId.isEmpty()) {
+            clientId = "mqtt-inbound-" + UUID.randomUUID();
+        }
+        
+        MqttPahoMessageDrivenChannelAdapter adapter = 
+                new MqttPahoMessageDrivenChannelAdapter(clientId + "-inbound", mqttClientFactory);
+        
+        // 设置消息转换器
+        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
+        converter.setPayloadAsBytes(false);
+        adapter.setConverter(converter);
+        
+        // 设置通道
+        adapter.setOutputChannel(mqttInputChannel());
+        
+        // 设置QoS
+        adapter.setQos(mqttProperties.getDefaultQos());
+        
+        return adapter;
+    }
+    
+    @Bean
+    @ConditionalOnMissingBean
+    public MqttSubscriberProcessor mqttSubscriberProcessor(MqttPahoMessageDrivenChannelAdapter mqttInbound,
+                                                           MqttClient mqttClient) {
+        return new MqttSubscriberProcessor(mqttInbound, mqttClient);
+    }
+    
+    /**
+     * MQTT客户端回调类
+     */
+    private static class MqttClientCallback implements org.eclipse.paho.client.mqttv3.MqttCallback {
+        private final MqttClient mqttClient;
+        private final List<MqttConnectionListener> connectionListeners;
+        
+        public MqttClientCallback(MqttClient mqttClient, List<MqttConnectionListener> connectionListeners) {
+            this.mqttClient = mqttClient;
+            this.connectionListeners = connectionListeners;
+        }
+        
+        @Override
+        public void connectionLost(Throwable cause) {
+            log.warn("MQTT connection lost: {}", cause.getMessage());
+            
+            // 通知连接监听器
+            for (MqttConnectionListener listener : connectionListeners) {
+                try {
+                    listener.connectionLost(cause);
+                } catch (Exception e) {
+                    log.error("Error notifying connection listener", e);
+                }
+            }
+        }
+        
+        @Override
+        public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) {
+            // 消息由MqttPahoMessageDrivenChannelAdapter处理
+        }
+        
+        @Override
+        public void deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token) {
+            // 通知连接监听器
+            for (MqttConnectionListener listener : connectionListeners) {
+                try {
+                    listener.deliveryComplete(token);
+                } catch (Exception e) {
+                    log.error("Error notifying delivery complete listener", e);
+                }
+            }
+        }
+    }
 } 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/config/MqttProperties.java

@@ -3,24 +3,168 @@ package cn.hfln.framework.mqtt.config;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
+/**
+ * MQTT配置属性
+ */
 @Data
 @ConfigurationProperties(prefix = "mqtt")
 public class MqttProperties {
+    /**
+     * 是否启用MQTT
+     */
     private boolean enabled = true;
-    private String broker;
+    
+    /**
+     * MQTT服务器地址
+     * 例如：tcp://localhost:1883
+     */
+    private String broker = "tcp://localhost:1883";
+    
+    /**
+     * 客户端ID
+     * 如果为空，会自动生成
+     */
     private String clientId;
+    
+    /**
+     * 用户名
+     */
     private String username;
+    
+    /**
+     * 密码
+     */
     private String password;
+    
+    /**
+     * 连接超时时间（秒）
+     */
     private int timeout = 30;
+    
+    /**
+     * 心跳间隔（秒）
+     */
     private int keepalive = 60;
+    
+    /**
+     * 是否清除会话
+     */
     private boolean cleanSession = true;
+    
+    /**
+     * 是否自动重连
+     */
     private boolean automaticReconnect = true;
-    private int maxReconnectDelay = 10000;
-    private int connectionTimeout = 30;
-    private int maxMessageSize = 1024 * 1024;
-    private boolean ssl = false;
-    private String sslCaPath;
-    private String sslClientCertPath;
-    private String sslClientKeyPath;
-    private String sslClientKeyPassword;
+    
+    /**
+     * 最大重连尝试次数
+     * 设置为-1表示无限重试
+     */
+    private int maxReconnectAttempts = 10;
+    
+    /**
+     * 重连间隔（毫秒）
+     */
+    private long reconnectInterval = 5000;
+    
+    /**
+     * 重连间隔增长因子
+     * 每次重连失败后，下次重连间隔会乘以这个因子
+     * 例如：初始间隔5秒，因子1.5，则重连间隔会变为：5, 7.5, 11.25, ...
+     */
+    private double reconnectBackoffMultiplier = 1.5;
+    
+    /**
+     * 最大重连间隔（毫秒）
+     * 重连间隔增长到这个值后不再增长
+     */
+    private long maxReconnectInterval = 60000;
+    
+    /**
+     * 是否在初始连接失败时抛出异常
+     * 如果为false，则初始连接失败时不会抛出异常，而是在后台进行重连
+     */
+    private boolean failFastOnInitialConnection = false;
+    
+    /**
+     * 连接恢复后是否自动重新订阅之前的主题
+     */
+    private boolean resubscribeOnReconnect = true;
+    
+    /**
+     * 默认QoS级别
+     */
+    private int defaultQos = 1;
+    
+    /**
+     * 默认是否保留消息
+     */
+    private boolean defaultRetained = false;
+    
+    /**
+     * 是否异步发送
+     */
+    private boolean async = true;
+    
+    /**
+     * 是否异步事件
+     */
+    private boolean asyncEvents = true;
+    
+    /**
+     * 默认主题
+     */
+    private String defaultTopic = "default";
+    
+    /**
+     * 订阅主题列表
+     */
+    private String[] subscribeTopics = new String[0];
+    
+    /**
+     * 订阅QoS级别
+     */
+    private int subscribeQos = 1;
+    
+    /**
+     * 完成超时时间（毫秒）
+     */
+    private long completionTimeout = 5000;
+    
+    /**
+     * 连接超时时间（秒）
+     */
+    public int getConnectionTimeout() {
+        return timeout;
+    }
+    
+    /**
+     * 发送消息失败时是否重试
+     */
+    private boolean retryOnPublishFailure = true;
+    
+    /**
+     * 发送消息失败时的最大重试次数
+     */
+    private int maxPublishRetryAttempts = 3;
+    
+    /**
+     * 发送消息重试间隔（毫秒）
+     */
+    private long publishRetryInterval = 1000;
+    
+    /**
+     * 是否启用连接健康检查
+     */
+    private boolean healthCheckEnabled = false;
+    
+    /**
+     * 健康检查间隔（毫秒）
+     */
+    private long healthCheckInterval = 30000;
+    
+    /**
+     * 健康检查主题
+     */
+    private String healthCheckTopic = "$SYS/healthcheck";
 }

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/connection/MqttReconnectManager.java

@@ -0,0 +1,292 @@
+package cn.hfln.framework.mqtt.connection;
+
+import cn.hfln.framework.mqtt.config.MqttProperties;
+import cn.hfln.framework.mqtt.listener.MqttConnectionListener;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * MQTT閲嶈繛绠＄悊鍣?
+ * 璐熻矗澶勭悊MQTT杩炴帴鐨勯噸杩為€昏緫鍜屽仴搴锋鏌?
+ */
+@Slf4j
+@Component
+public class MqttReconnectManager implements DisposableBean {
+
+    private final MqttClient mqttClient;
+    private final MqttProperties mqttProperties;
+    private final MqttConnectOptions mqttConnectOptions;
+    private final List<MqttConnectionListener> connectionListeners;
+    
+    private final ScheduledExecutorService scheduler;
+    private final AtomicBoolean reconnecting = new AtomicBoolean(false);
+    private final AtomicInteger reconnectAttempts = new AtomicInteger(0);
+    private long currentReconnectInterval;
+    
+    // 淇濆瓨璁㈤槄鐨勪富棰樺拰QoS
+    private final Map<String, Integer> subscribedTopics = new ConcurrentHashMap<>();
+    
+    @Autowired
+    public MqttReconnectManager(MqttClient mqttClient, 
+                               MqttProperties mqttProperties,
+                               MqttConnectOptions mqttConnectOptions,
+                               List<MqttConnectionListener> connectionListeners) {
+        this.mqttClient = mqttClient;
+        this.mqttProperties = mqttProperties;
+        this.mqttConnectOptions = mqttConnectOptions;
+        this.connectionListeners = connectionListeners;
+        this.currentReconnectInterval = mqttProperties.getReconnectInterval();
+        this.scheduler = Executors.newScheduledThreadPool(1, r -> {
+            Thread thread = new Thread(r, "mqtt-reconnect-thread");
+            thread.setDaemon(true);
+            return thread;
+        });
+        
+        // 鍚姩鍋ュ悍妫€鏌?
+        if (mqttProperties.isHealthCheckEnabled()) {
+            startHealthCheck();
+        }
+    }
+    
+    /**
+     * 澶勭悊杩炴帴涓㈠け浜嬩欢
+     * 
+     * @param cause 杩炴帴涓㈠け鍘熷洜
+     */
+    public void handleConnectionLost(Throwable cause) {
+        log.warn("MQTT connection lost: {}", cause.getMessage());
+        
+        // 閫氱煡杩炴帴鐩戝惉鍣?
+        for (MqttConnectionListener listener : connectionListeners) {
+            try {
+                listener.connectionLost(cause);
+            } catch (Exception e) {
+                log.error("Error notifying connection listener", e);
+            }
+        }
+        
+        // 濡傛灉鍚敤浜嗚嚜鍔ㄩ噸杩烇紝鍒欏紑濮嬮噸杩?
+        if (mqttProperties.isAutomaticReconnect() && !reconnecting.getAndSet(true)) {
+            startReconnect();
+        }
+    }
+    
+    /**
+     * 璁板綍璁㈤槄鐨勪富棰?
+     * 
+     * @param topic 涓婚
+     * @param qos QoS绾у埆
+     */
+    public void recordSubscription(String topic, int qos) {
+        subscribedTopics.put(topic, qos);
+    }
+    
+    /**
+     * 绉婚櫎璁㈤槄鐨勪富棰?
+     * 
+     * @param topic 涓婚
+     */
+    public void removeSubscription(String topic) {
+        subscribedTopics.remove(topic);
+    }
+    
+    /**
+     * 寮€濮嬮噸杩?
+     */
+    private void startReconnect() {
+        reconnectAttempts.set(0);
+        currentReconnectInterval = mqttProperties.getReconnectInterval();
+        scheduleReconnect();
+    }
+    
+    /**
+     * 璋冨害閲嶈繛浠诲姟
+     */
+    private void scheduleReconnect() {
+        int maxAttempts = mqttProperties.getMaxReconnectAttempts();
+        if (maxAttempts > 0 && reconnectAttempts.get() >= maxAttempts) {
+            log.error("Maximum reconnect attempts ({}) reached, giving up reconnecting to MQTT broker", maxAttempts);
+            reconnecting.set(false);
+            
+            // 閫氱煡杩炴帴鐩戝惉鍣?
+            for (MqttConnectionListener listener : connectionListeners) {
+                try {
+                    listener.onConnectionFailed(new Exception("Maximum reconnect attempts reached"));
+                } catch (Exception e) {
+                    log.error("Error notifying connection listener", e);
+                }
+            }
+            return;
+        }
+        
+        log.info("Scheduling reconnect attempt in {} ms", currentReconnectInterval);
+        scheduler.schedule(this::attemptReconnect, currentReconnectInterval, TimeUnit.MILLISECONDS);
+    }
+    
+    /**
+     * 灏濊瘯閲嶈繛
+     */
+    private void attemptReconnect() {
+        try {
+            int attempt = reconnectAttempts.incrementAndGet();
+            log.info("Attempting to reconnect to MQTT broker, attempt: {}/{}", 
+                    attempt, mqttProperties.getMaxReconnectAttempts() > 0 ? 
+                            mqttProperties.getMaxReconnectAttempts() : "unlimited");
+            
+            if (!mqttClient.isConnected()) {
+                mqttClient.connect(mqttConnectOptions);
+                log.info("Successfully reconnected to MQTT broker");
+                
+                // 閲嶆柊璁㈤槄涔嬪墠鐨勪富棰?
+                if (mqttProperties.isResubscribeOnReconnect()) {
+                    resubscribe();
+                }
+                
+                // 閫氱煡杩炴帴鐩戝惉鍣?
+                for (MqttConnectionListener listener : connectionListeners) {
+                    try {
+                        listener.onConnected();
+                    } catch (Exception e) {
+                        log.error("Error notifying connection listener", e);
+                    }
+                }
+                
+                reconnecting.set(false);
+            }
+        } catch (MqttException e) {
+            log.error("Failed to reconnect to MQTT broker: {}, will retry", e.getMessage());
+            
+            // 璁＄畻涓嬩竴娆￠噸杩為棿闅?
+            calculateNextReconnectInterval();
+            
+            // 璋冨害涓嬩竴娆￠噸杩?
+            scheduleReconnect();
+        }
+    }
+    
+    /**
+     * 璁＄畻涓嬩竴娆￠噸杩為棿闅?
+     */
+    private void calculateNextReconnectInterval() {
+        // 浣跨敤鎸囨暟閫€閬跨畻娉?
+        currentReconnectInterval = Math.min(
+            (long)(currentReconnectInterval * mqttProperties.getReconnectBackoffMultiplier()),
+            mqttProperties.getMaxReconnectInterval()
+        );
+        log.debug("Next reconnect interval: {} ms", currentReconnectInterval);
+    }
+    
+    /**
+     * 閲嶆柊璁㈤槄涔嬪墠鐨勪富棰?
+     */
+    private void resubscribe() {
+        if (subscribedTopics.isEmpty()) {
+            return;
+        }
+        
+        log.info("Resubscribing to {} topics", subscribedTopics.size());
+        
+        subscribedTopics.forEach((topic, qos) -> {
+            try {
+                mqttClient.subscribe(topic, qos);
+                log.debug("Resubscribed to topic: {} with QoS: {}", topic, qos);
+            } catch (MqttException e) {
+                log.error("Failed to resubscribe to topic: {}", topic, e);
+            }
+        });
+    }
+    
+    /**
+     * 鍚姩鍋ュ悍妫€鏌?
+     */
+    private void startHealthCheck() {
+        log.info("Starting MQTT connection health check with interval: {} ms", 
+                mqttProperties.getHealthCheckInterval());
+        
+        scheduler.scheduleAtFixedRate(
+            this::performHealthCheck,
+            mqttProperties.getHealthCheckInterval(),
+            mqttProperties.getHealthCheckInterval(),
+            TimeUnit.MILLISECONDS
+        );
+    }
+    
+    /**
+     * 鎵ц鍋ュ悍妫€鏌?
+     */
+    private void performHealthCheck() {
+        try {
+            if (!mqttClient.isConnected() && !reconnecting.get()) {
+                log.warn("Health check detected disconnected MQTT client, initiating reconnect");
+                handleConnectionLost(new Exception("Health check detected disconnection"));
+            } else {
+                // 鍙互鍦ㄨ繖閲屾坊鍔犻澶栫殑鍋ュ悍妫€鏌ラ€昏緫锛屼緥濡傚彂閫乸ing娑堟伅
+                log.debug("Health check: MQTT connection is {}", 
+                        mqttClient.isConnected() ? "active" : "inactive but reconnecting");
+            }
+        } catch (Exception e) {
+            log.error("Error during MQTT health check", e);
+        }
+    }
+    
+    /**
+     * 鎵嬪姩瑙﹀彂閲嶈繛
+     */
+    public void triggerReconnect() {
+        if (!reconnecting.getAndSet(true)) {
+            log.info("Manually triggering MQTT reconnection");
+            startReconnect();
+        } else {
+            log.debug("Reconnection already in progress, ignoring trigger request");
+        }
+    }
+    
+    /**
+     * 鑾峰彇褰撳墠杩炴帴鐘舵€?
+     * 
+     * @return 杩炴帴鐘舵€佷俊鎭?
+     */
+    public Map<String, Object> getConnectionStatus() {
+        Map<String, Object> status = new HashMap<>();
+        status.put("connected", mqttClient.isConnected());
+        status.put("clientId", mqttClient.getClientId());
+        status.put("serverURI", mqttClient.getServerURI());
+        status.put("reconnecting", reconnecting.get());
+        status.put("reconnectAttempts", reconnectAttempts.get());
+        status.put("subscribedTopics", subscribedTopics.size());
+        return status;
+    }
+    
+    /**
+     * 閿€姣佽祫婧?
+     */
+    @Override
+    public void destroy() {
+        log.info("Shutting down MQTT reconnect manager");
+        scheduler.shutdownNow();
+        try {
+            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+                log.warn("Scheduler did not terminate in time");
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/converter/JsonMessageConverter.java

@@ -1,38 +1,80 @@
 package cn.hfln.framework.mqtt.converter;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.stereotype.Component;
 
-import java.nio.charset.StandardCharsets;
-
+/**
+ * JSON娑堟伅杞崲鍣?
+ * 鐢ㄤ簬灏嗗璞¤浆鎹负JSON瀛楃涓诧紝鎴栧皢JSON瀛楃涓茶浆鎹负瀵硅薄
+ */
 @Slf4j
+@Component
 public class JsonMessageConverter implements MessageConverter {
-    private final ObjectMapper objectMapper;
 
-    public JsonMessageConverter() {
-        this.objectMapper = new ObjectMapper();
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    /**
+     * 灏嗗璞¤浆鎹负JSON瀛楃涓?
+     * 
+     * @param payload 瀵硅薄
+     * @return JSON瀛楃涓?
+     * @throws JsonProcessingException JSON澶勭悊寮傚父
+     */
+    public String toJson(Object payload) throws JsonProcessingException {
+        if (payload == null) {
+            return null;
+        }
+        
+        if (payload instanceof String) {
+            return (String) payload;
+        }
+        
+        return objectMapper.writeValueAsString(payload);
     }
 
+    /**
+     * 灏咼SON瀛楃涓茶浆鎹负瀵硅薄
+     * 
+     * @param <T> 鐩爣绫诲瀷
+     * @param json JSON瀛楃涓?
+     * @param clazz 鐩爣绫诲瀷
+     * @return 杞崲鍚庣殑瀵硅薄
+     * @throws JsonProcessingException JSON澶勭悊寮傚父
+     */
+    public <T> T fromJson(String json, Class<T> clazz) throws JsonProcessingException {
+        if (json == null || json.isEmpty()) {
+            return null;
+        }
+        
+        return objectMapper.readValue(json, clazz);
+    }
+    
     @Override
     public MqttMessage toMessage(Object payload) {
         try {
-            String json = objectMapper.writeValueAsString(payload);
-            return new MqttMessage(json.getBytes(StandardCharsets.UTF_8));
-        } catch (Exception e) {
-            log.error("Failed to convert object to message", e);
-            throw new RuntimeException("Failed to convert object to message", e);
+            String json = toJson(payload);
+            MqttMessage message = new MqttMessage();
+            message.setPayload(json == null ? "null".getBytes() : json.getBytes());
+            return message;
+        } catch (JsonProcessingException e) {
+            log.error("Error converting object to MQTT message", e);
+            throw new RuntimeException("Error converting object to MQTT message", e);
         }
     }
 
     @Override
     public <T> T fromMessage(MqttMessage message, Class<T> targetType) {
         try {
-            String json = new String(message.getPayload(), StandardCharsets.UTF_8);
-            return objectMapper.readValue(json, targetType);
-        } catch (Exception e) {
-            log.error("Failed to convert message to object", e);
-            throw new RuntimeException("Failed to convert message to object", e);
+            String json = new String(message.getPayload());
+            return fromJson(json, targetType);
+        } catch (JsonProcessingException e) {
+            log.error("Error converting MQTT message to object", e);
+            throw new RuntimeException("Error converting MQTT message to object", e);
         }
     }
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/converter/MessageConverter.java

@@ -22,4 +22,6 @@ public interface MessageConverter {
      * @return 转换后的对象
      */
     <T> T fromMessage(MqttMessage message, Class<T> targetType);
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/exception/MqttExceptionHandler.java

@@ -0,0 +1,29 @@
+package cn.hfln.framework.mqtt.exception;
+
+import org.eclipse.paho.client.mqttv3.MqttException;
+
+import java.util.function.Consumer;
+
+public class MqttExceptionHandler {
+    public static void handleException(MqttException ex, Consumer<MqttException> recoverAction) {
+        // 默认空实现
+        if (recoverAction != null) {
+            recoverAction.accept(ex);
+        }
+    }
+
+    public static boolean isRecoverableByReconnect(MqttException ex) {
+        // 默认全部不可恢复
+        return false;
+    }
+
+    public static String getDetailedErrorMessage(MqttException ex) {
+        // 返回异常信息
+        return ex != null ? ex.getMessage() : "";
+    }
+
+    public static String getRecommendedAction(MqttException ex) {
+        // 返回默认建议
+        return "请检查MQTT连接配置或网络状态";
+    }
+}

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/gateway/DefaultMqttGateway.java

@@ -0,0 +1,224 @@
+package cn.hfln.framework.mqtt.gateway;
+
+import cn.hfln.framework.mqtt.template.MqttTemplate;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * MQTT网关默认实现
+ * 使用MqttTemplate实现MQTT消息发送功能
+ */
+@Slf4j
+@Component
+public class DefaultMqttGateway implements MqttGateway {
+
+    private final MqttTemplate mqttTemplate;
+    private final MqttClient mqttClient;
+
+    @Autowired
+    public DefaultMqttGateway(MqttTemplate mqttTemplate, MqttClient mqttClient) {
+        this.mqttTemplate = mqttTemplate;
+        this.mqttClient = mqttClient;
+    }
+
+    @Override
+    public void initialize() {
+        log.info("MQTT Gateway initialized");
+    }
+
+    @Override
+    public void publish(String topic, Object payload) {
+        try {
+            if (payload instanceof String) {
+                mqttTemplate.send(topic, (String) payload);
+            } else {
+                mqttTemplate.sendJson(topic, payload);
+            }
+            log.debug("Message published to topic: {}", topic);
+        } catch (Exception e) {
+            log.error("Error publishing message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void publish(String topic, Object payload, int qos, boolean retain) {
+        try {
+            if (payload instanceof String) {
+                mqttTemplate.send(topic, (String) payload, qos, retain);
+            } else {
+                mqttTemplate.sendJson(topic, payload, qos, retain);
+            }
+            log.debug("Message published to topic: {} with QoS: {}, retain: {}", topic, qos, retain);
+        } catch (Exception e) {
+            log.error("Error publishing message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void sendMessage(String topic, String message) {
+        try {
+            mqttTemplate.send(topic, message);
+            log.debug("Message sent to topic: {}", topic);
+        } catch (Exception e) {
+            log.error("Error sending message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void sendMessage(String topic, String message, int qos, boolean retain) {
+        try {
+            mqttTemplate.send(topic, message, qos, retain);
+            log.debug("Message sent to topic: {} with QoS: {}, retain: {}", topic, qos, retain);
+        } catch (Exception e) {
+            log.error("Error sending message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void publishJson(String topic, Object payload) {
+        try {
+            mqttTemplate.sendJson(topic, payload);
+            log.debug("JSON message published to topic: {}", topic);
+        } catch (Exception e) {
+            log.error("Error publishing JSON message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void publishJson(String topic, Object payload, int qos, boolean retain) {
+        try {
+            mqttTemplate.sendJson(topic, payload, qos, retain);
+            log.debug("JSON message published to topic: {} with QoS: {}, retain: {}", topic, qos, retain);
+        } catch (Exception e) {
+            log.error("Error publishing JSON message to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void sendSync(String topic, Object payload) throws Exception {
+        try {
+            if (payload instanceof String) {
+                mqttTemplate.sendSync(topic, (String) payload);
+            } else {
+                mqttTemplate.sendJsonSync(topic, payload);
+            }
+            log.debug("Message sent synchronously to topic: {}", topic);
+        } catch (MqttException e) {
+            log.error("Error sending synchronous message to topic: {}", topic, e);
+            throw e;
+        }
+    }
+
+    @Override
+    public void subscribe(String topic, int qos) {
+        try {
+            mqttClient.subscribe(topic, qos);
+            log.info("Subscribed to topic: {} with QoS: {}", topic, qos);
+        } catch (MqttException e) {
+            log.error("Error subscribing to topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public void unsubscribe(String topic) {
+        try {
+            mqttClient.unsubscribe(topic);
+            log.info("Unsubscribed from topic: {}", topic);
+        } catch (MqttException e) {
+            log.error("Error unsubscribing from topic: {}", topic, e);
+        }
+    }
+
+    @Override
+    public boolean isConnected() {
+        return mqttClient.isConnected();
+    }
+
+    @Override
+    public void disconnect() {
+        try {
+            mqttClient.disconnect();
+            log.info("MQTT client disconnected");
+        } catch (MqttException e) {
+            log.error("Error disconnecting MQTT client", e);
+        }
+    }
+    
+    /**
+     * 发送通用消息
+     * 
+     * @param topic 主题
+     * @param messageType 消息类型
+     * @param messageData 消息数据
+     */
+    @Override
+    public void sendGenericMessage(String topic, String messageType, Map<String, Object> messageData) {
+        try {
+            Map<String, Object> payload = new HashMap<>();
+            payload.put("message_type", messageType);
+            payload.put("timestamp", System.currentTimeMillis());
+            
+            if (messageData != null) {
+                payload.putAll(messageData);
+            }
+            
+            publishJson(topic, payload);
+        } catch (Exception e) {
+            log.error("Error sending generic message to topic: {}", topic, e);
+        }
+    }
+    
+    /**
+     * 发送命令消息
+     * 
+     * @param topic 主题
+     * @param command 命令
+     * @param params 参数
+     */
+    @Override
+    public void sendCommand(String topic, String command, Map<String, Object> params) {
+        try {
+            Map<String, Object> payload = new HashMap<>();
+            payload.put("command", command);
+            payload.put("timestamp", System.currentTimeMillis());
+            
+            if (params != null) {
+                payload.put("params", params);
+            }
+            
+            publishJson(topic, payload);
+        } catch (Exception e) {
+            log.error("Error sending command to topic: {}", topic, e);
+        }
+    }
+    
+    /**
+     * 发送响应消息
+     * 
+     * @param topic 主题
+     * @param code 响应码
+     * @param data 响应数据
+     */
+    @Override
+    public void sendResponse(String topic, int code, Map<String, Object> data) {
+        try {
+            Map<String, Object> payload = new HashMap<>();
+            payload.put("code", code);
+            payload.put("timestamp", System.currentTimeMillis());
+            
+            if (data != null) {
+                payload.putAll(data);
+            }
+            
+            publishJson(topic, payload);
+        } catch (Exception e) {
+            log.error("Error sending response to topic: {}", topic, e);
+        }
+    }
+} 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/gateway/MqttGateway.java

@@ -0,0 +1,132 @@
+package cn.hfln.framework.mqtt.gateway;
+
+import java.util.Map;
+
+/**
+ * MQTT网关接口
+ * 定义MQTT消息发送的基本功能
+ */
+public interface MqttGateway {
+
+    /**
+     * 初始化MQTT客户端
+     */
+    void initialize();
+    
+    /**
+     * 发布MQTT消息
+     * 
+     * @param topic 主题
+     * @param payload 消息内容
+     */
+    void publish(String topic, Object payload);
+    
+    /**
+     * 发布MQTT消息（指定QoS和保留标志）
+     * 
+     * @param topic 主题
+     * @param payload 消息内容
+     * @param qos QoS级别(0,1,2)
+     * @param retain 是否保留消息
+     */
+    void publish(String topic, Object payload, int qos, boolean retain);
+    
+    /**
+     * 发布MQTT消息（字符串内容）
+     * 
+     * @param topic 主题
+     * @param message 消息内容
+     */
+    void sendMessage(String topic, String message);
+    
+    /**
+     * 发布MQTT消息（字符串内容，指定QoS和保留标志）
+     * 
+     * @param topic 主题
+     * @param message 消息内容
+     * @param qos QoS级别(0,1,2)
+     * @param retain 是否保留消息
+     */
+    void sendMessage(String topic, String message, int qos, boolean retain);
+    
+    /**
+     * 发布JSON格式的MQTT消息
+     * 
+     * @param topic 主题
+     * @param payload 消息内容对象
+     */
+    void publishJson(String topic, Object payload);
+    
+    /**
+     * 发布JSON格式的MQTT消息（指定QoS和保留标志）
+     * 
+     * @param topic 主题
+     * @param payload 消息内容对象
+     * @param qos QoS级别(0,1,2)
+     * @param retain 是否保留消息
+     */
+    void publishJson(String topic, Object payload, int qos, boolean retain);
+    
+    /**
+     * 同步发送消息
+     * 
+     * @param topic 主题
+     * @param payload 消息内容
+     * @throws Exception 发送异常
+     */
+    void sendSync(String topic, Object payload) throws Exception;
+    
+    /**
+     * 订阅主题
+     * 
+     * @param topic 主题
+     * @param qos QoS级别
+     */
+    void subscribe(String topic, int qos);
+    
+    /**
+     * 取消订阅主题
+     * 
+     * @param topic 主题
+     */
+    void unsubscribe(String topic);
+    
+    /**
+     * 判断是否已连接
+     * 
+     * @return 是否已连接
+     */
+    boolean isConnected();
+    
+    /**
+     * 关闭连接
+     */
+    void disconnect();
+    
+    /**
+     * 发送通用消息
+     * 
+     * @param topic 主题
+     * @param messageType 消息类型
+     * @param messageData 消息数据
+     */
+    void sendGenericMessage(String topic, String messageType, Map<String, Object> messageData);
+    
+    /**
+     * 发送命令消息
+     * 
+     * @param topic 主题
+     * @param command 命令
+     * @param params 参数
+     */
+    void sendCommand(String topic, String command, Map<String, Object> params);
+    
+    /**
+     * 发送响应消息
+     * 
+     * @param topic 主题
+     * @param code 响应码
+     * @param data 响应数据
+     */
+    void sendResponse(String topic, int code, Map<String, Object> data);
+} 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/GenericMqttMessageHandler.java

@@ -0,0 +1 @@
+package cn.hfln.framework.mqtt.handler;

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/MqttMessageHandler.java

@@ -1,89 +1,22 @@
 package cn.hfln.framework.mqtt.handler;
 
-import cn.hfln.framework.mqtt.converter.MessageConverter;
-import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
-import org.springframework.stereotype.Component;
 
-import java.lang.reflect.Method;
-import java.nio.charset.StandardCharsets;
-
-@Component
-@Slf4j
 public class MqttMessageHandler implements MqttCallback {
-
-    @Autowired
-    private ApplicationContext applicationContext;
-
-    @Autowired
-    private MqttSubscribeProcessor subscribeProcessor;
-
-    @Autowired
-    private MessageConverter messageConverter;
-
     @Override
     public void connectionLost(Throwable cause) {
-        log.error("Connection lost", cause);
+        // 连接丢失时的处理
     }
 
     @Override
     public void messageArrived(String topic, MqttMessage message) {
-//        log.info("Message arrived on topic: {}", topic);
-//        log.debug("Message content: {}", new String(message.getPayload()));
-
-        Method method = subscribeProcessor.getMethodForTopic(topic);
-        if (method == null) {
-            log.warn("No handler method found for topic: {}", topic);
-            return;
-        }
-
-        try {
-            Object bean = applicationContext.getBean(method.getDeclaringClass());
-//            log.debug("Found handler bean: {} for topic: {}", bean.getClass().getName(), topic);
-            
-            // 获取方法参数类型
-            Class<?>[] parameterTypes = method.getParameterTypes();
-            if (parameterTypes.length != 1) {
-                log.error("Handler method must have exactly one parameter, but found {} parameters", parameterTypes.length);
-                return;
-            }
-
-            Object[] args = new Object[1];
-            Class<?> paramType = parameterTypes[0];
-            
-            // 根据参数类型转换消息内容
-            if (paramType == String.class) {
-                args[0] = new String(message.getPayload(), StandardCharsets.UTF_8);
-//                log.debug("Converting message to String: {}", args[0]);
-            } else if (paramType == byte[].class) {
-                args[0] = message.getPayload();
-//                log.debug("Using raw byte array as message payload");
-            } else {
-                // 使用消息转换器转换
-                args[0] = messageConverter.fromMessage(message, paramType);
-//                log.debug("Converting message to {}: {}", paramType.getName(), args[0]);
-            }
-            
-//            log.debug("Invoking method: {} on bean: {} with args: {}", method.getName(), bean.getClass().getName(), args[0]);
-            Object result = method.invoke(bean, args);
-//            log.debug("Method invocation result: {}", result);
-//            log.info("Successfully processed message for topic: {}", topic);
-        } catch (Exception e) {
-            log.error("Error invoking method for topic: {} - Error: {}", topic, e.getMessage(), e);
-        }
+        // 收到消息时的处理
     }
 
     @Override
     public void deliveryComplete(IMqttDeliveryToken token) {
-        try {
-//            log.debug("Message delivery complete for token: {}", token);
-//            log.debug("Message delivered to topics: {}", (Object) token.getTopics());
-        } catch (Exception e) {
-            log.error("Error getting delivery topics", e);
-        }
+        // 消息发送完成时的处理
     }
-} 
+}

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/MqttMessageRouter.java

@@ -0,0 +1,102 @@
+package cn.hfln.framework.mqtt.handler;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;
+
+/**
+ * MQTT消息路由器
+ * 根据主题模式匹配将消息路由到对应的处理器
+ */
+@Slf4j
+@Component
+public class MqttMessageRouter {
+
+    // 主题模式和处理器的映射关系
+    private final Map<Pattern, MqttTopicHandler> handlerMap = new ConcurrentHashMap<>();
+
+    /**
+     * 注册主题处理器
+     * @param topicPattern 主题模式（正则表达式）
+     * @param handler 处理器
+     * @return 路由器实例，支持链式调用
+     */
+    public MqttMessageRouter register(String topicPattern, MqttTopicHandler handler) {
+        Pattern pattern = Pattern.compile(topicPattern);
+        handlerMap.put(pattern, handler);
+        log.info("Registered MQTT topic handler for pattern: {}", topicPattern);
+        return this;
+    }
+
+    /**
+     * 注册主题处理器
+     * @param pattern 主题模式（已编译的正则表达式）
+     * @param handler 处理器
+     * @return 路由器实例，支持链式调用
+     */
+    public MqttMessageRouter register(Pattern pattern, MqttTopicHandler handler) {
+        handlerMap.put(pattern, handler);
+        log.info("Registered MQTT topic handler for pattern: {}", pattern.pattern());
+        return this;
+    }
+
+    /**
+     * 取消注册主题处理器
+     * @param topicPattern 主题模式（正则表达式）
+     * @return 路由器实例，支持链式调用
+     */
+    public MqttMessageRouter unregister(String topicPattern) {
+        for (Pattern pattern : handlerMap.keySet()) {
+            if (pattern.pattern().equals(topicPattern)) {
+                handlerMap.remove(pattern);
+                log.info("Unregistered MQTT topic handler for pattern: {}", topicPattern);
+                break;
+            }
+        }
+        return this;
+    }
+
+    /**
+     * 路由消息到相应的处理器
+     * @param topic 消息主题
+     * @param payload 消息内容
+     * @return 是否找到匹配的处理器
+     */
+    public boolean route(String topic, String payload) {
+        boolean handled = false;
+        
+        for (Map.Entry<Pattern, MqttTopicHandler> entry : handlerMap.entrySet()) {
+            Pattern pattern = entry.getKey();
+            MqttTopicHandler handler = entry.getValue();
+            
+            if (pattern.matcher(topic).matches()) {
+                try {
+                    handler.handle(topic, payload);
+                    handled = true;
+                    log.debug("Routed MQTT message to handler: topic={}, pattern={}", topic, pattern.pattern());
+                } catch (Exception e) {
+                    log.error("Error handling MQTT message: topic={}, pattern={}", topic, pattern.pattern(), e);
+                }
+            }
+        }
+        
+        if (!handled) {
+            log.debug("No handler found for MQTT topic: {}", topic);
+        }
+        
+        return handled;
+    }
+
+    /**
+     * 清除所有已注册的处理器
+     */
+    public void clear() {
+        handlerMap.clear();
+        log.info("Cleared all MQTT topic handlers");
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/MqttSubscribeProcessor.java

@@ -41,7 +41,7 @@ public class MqttSubscribeProcessor implements BeanPostProcessor {
             processMqttSubscribe(bean, beanClass, classAnnotation);
         }
 
-        // 处理方法级别的注解
+        // 处理方法级别的订阅
         ReflectionUtils.doWithMethods(beanClass, method -> {
             MqttSubscribe annotation = method.getAnnotation(MqttSubscribe.class);
             if (annotation != null) {
@@ -68,7 +68,8 @@ public class MqttSubscribeProcessor implements BeanPostProcessor {
             if (target instanceof Method) {
                 topicMethodMap.put(topic, (Method) target);
             } else if (target instanceof Class) {
-                // 对于类级别的注解，我们需要找到合适的处理方法
+                // 对于类级别的注解，需要找到合适的处理方法
+                // 首先尝试精确匹配
                 Method method = findHandlerMethod((Class<?>) target);
                 if (method != null) {
                     topicMethodMap.put(topic, method);
@@ -93,17 +94,18 @@ public class MqttSubscribeProcessor implements BeanPostProcessor {
     }
 
     public Method getMethodForTopic(String topic) {
+        // 对于类级别的注解，需要找到合适的处理方法
         // 首先尝试精确匹配
         Method method = topicMethodMap.get(topic);
         if (method != null) {
             return method;
         }
 
-        // 如果没有精确匹配，尝试通配符匹配
+        // 如果没有精确匹配，则尝试通配符匹配
         for (Map.Entry<String, Method> entry : topicMethodMap.entrySet()) {
             String pattern = entry.getKey()
-                    .replace("+", "[^/]+")  // 将 + 替换为匹配除 / 外的任意字符
-                    .replace("#", ".*");     // 将 # 替换为匹配任意字符
+                    .replace("+", "[^/]+")  // 对于 + ，需要替换为 [^/]+ ，以匹配除 / 之外的任何字符
+                    .replace("#", ".*");     // 对于 # ，需要替换为 .* ，以匹配任何字符串
             if (Pattern.matches(pattern, topic)) {
 //                log.info("Found matching method for topic: {} using pattern: {}", topic, pattern);
                 return entry.getValue();
@@ -113,4 +115,6 @@ public class MqttSubscribeProcessor implements BeanPostProcessor {
         log.warn("No method found for topic: {}", topic);
         return null;
     }
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/MqttSubscriberProcessor.java

@@ -0,0 +1,241 @@
+package cn.hfln.framework.mqtt.handler;
+
+import cn.hfln.framework.mqtt.annotation.MqttSubscriber;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.stereotype.Component;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * MQTT订阅处理器
+ * 扫描并处理带有@MqttSubscriber注解的方法
+ */
+@Slf4j
+@Component
+public class MqttSubscriberProcessor implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean {
+    
+    private ApplicationContext applicationContext;
+    private final MqttPahoMessageDrivenChannelAdapter mqttInbound;
+    private final MqttClient mqttClient;
+    private final MessageHandler messageHandler;
+    
+    // 保存主题和订阅方法的映射关系
+    private final Map<String, SubscriberMethod> topicSubscriberMap = new ConcurrentHashMap<>();
+    
+    public MqttSubscriberProcessor(MqttPahoMessageDrivenChannelAdapter mqttInbound, MqttClient mqttClient) {
+        this.mqttInbound = mqttInbound;
+        this.mqttClient = mqttClient;
+        
+        // 创建消息处理器
+        this.messageHandler = message -> {
+            try {
+                processMessage(message);
+            } catch (Exception e) {
+                log.error("Error processing MQTT message: {}", e.getMessage(), e);
+            }
+        };
+    }
+    
+    private void processMessage(Message<?> message) {
+        String receivedTopic = (String) message.getHeaders().get("mqtt_receivedTopic");
+        if (receivedTopic == null) {
+            log.warn("Received MQTT message without topic header");
+            return;
+        }
+        
+        // 查找对应的订阅方法
+        SubscriberMethod subscriberMethod = findSubscriberMethod(receivedTopic);
+        if (subscriberMethod == null) {
+            log.debug("No subscriber found for topic: {}", receivedTopic);
+            return;
+        }
+        
+        try {
+            // 调用订阅方法
+            invokeSubscriberMethod(subscriberMethod, receivedTopic, message);
+        } catch (Exception e) {
+            log.error("Error invoking MQTT subscriber method: {}", e.getMessage(), e);
+        }
+    }
+    
+    private SubscriberMethod findSubscriberMethod(String receivedTopic) {
+        // 精确匹配
+        if (topicSubscriberMap.containsKey(receivedTopic)) {
+            return topicSubscriberMap.get(receivedTopic);
+        }
+        
+        // 模式匹配
+        for (Map.Entry<String, SubscriberMethod> entry : topicSubscriberMap.entrySet()) {
+            if (matchTopic(entry.getKey(), receivedTopic)) {
+                return entry.getValue();
+            }
+        }
+        
+        return null;
+    }
+    
+    /**
+     * 匹配主题
+     * 支持通配符 + 和 #
+     * + 匹配一个层级
+     * # 匹配多个层级
+     */
+    private boolean matchTopic(String pattern, String topic) {
+        String[] patternSegments = pattern.split("/");
+        String[] topicSegments = topic.split("/");
+        
+        if (patternSegments.length > topicSegments.length) {
+            return false;
+        }
+        
+        for (int i = 0; i < patternSegments.length; i++) {
+            if (i == patternSegments.length - 1 && patternSegments[i].equals("#")) {
+                return true;
+            }
+            
+            if (!patternSegments[i].equals("+") && !patternSegments[i].equals(topicSegments[i])) {
+                return false;
+            }
+        }
+        
+        return patternSegments.length == topicSegments.length;
+    }
+    
+    private void invokeSubscriberMethod(SubscriberMethod subscriberMethod, String topic, Message<?> message) throws Exception {
+        Method method = subscriberMethod.getMethod();
+        Object bean = subscriberMethod.getBean();
+        
+        if (method.getParameterCount() == 1) {
+            // 一个参数：消息内容
+            Object payload = message.getPayload();
+            method.invoke(bean, payload);
+        } else if (method.getParameterCount() == 2) {
+            // 两个参数：主题和消息内容
+            Object payload = message.getPayload();
+            method.invoke(bean, topic, payload);
+        } else {
+            log.warn("Invalid subscriber method parameter count: {}", method.getParameterCount());
+        }
+    }
+    
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = applicationContext;
+    }
+    
+    @Override
+    public void afterSingletonsInstantiated() {
+        // 扫描所有Bean，查找带有@MqttSubscriber注解的方法
+        String[] beanNames = applicationContext.getBeanDefinitionNames();
+        for (String beanName : beanNames) {
+            Object bean = applicationContext.getBean(beanName);
+            scanSubscriberMethods(bean);
+        }
+        
+        // 设置消息通道
+        mqttInbound.setOutputChannelName("mqttInputChannel");
+    }
+    
+    private void scanSubscriberMethods(Object bean) {
+        Class<?> beanClass = bean.getClass();
+        for (Method method : beanClass.getMethods()) {
+            MqttSubscriber mqttSubscriber = method.getAnnotation(MqttSubscriber.class);
+            if (mqttSubscriber != null) {
+                registerSubscriberMethod(bean, method, mqttSubscriber);
+            }
+        }
+    }
+    
+    private void registerSubscriberMethod(Object bean, Method method, MqttSubscriber mqttSubscriber) {
+        // 检查方法参数
+        if (method.getParameterCount() != 1 && method.getParameterCount() != 2) {
+            log.error("MQTT subscriber method should have 1 or 2 parameters: {}.{}",
+                    bean.getClass().getName(), method.getName());
+            return;
+        }
+        
+        if (method.getParameterCount() == 2 && !method.getParameterTypes()[0].equals(String.class)) {
+            log.error("First parameter of MQTT subscriber method should be String (topic): {}.{}",
+                    bean.getClass().getName(), method.getName());
+            return;
+        }
+        
+        // 注册订阅方法
+        SubscriberMethod subscriberMethod = new SubscriberMethod(bean, method, mqttSubscriber);
+        topicSubscriberMap.put(mqttSubscriber.topic(), subscriberMethod);
+        
+        // 添加订阅
+        String formattedTopic = formatTopic(mqttSubscriber);
+        try {
+            log.info("Subscribe to MQTT topic: {}, QoS: {}", formattedTopic, mqttSubscriber.qos());
+            
+            // 添加到MQTT入站通道
+            mqttInbound.addTopic(formattedTopic, mqttSubscriber.qos());
+        } catch (Exception e) {
+            log.error("Error subscribing to topic: {}", formattedTopic, e);
+        }
+        
+        log.info("Registered MQTT subscriber: {}.{} for topic: {}",
+                bean.getClass().getName(), method.getName(), mqttSubscriber.topic());
+    }
+    
+    private String formatTopic(MqttSubscriber mqttSubscriber) {
+        if (mqttSubscriber.shared()) {
+            return "$share/" + mqttSubscriber.group() + "/" + mqttSubscriber.topic();
+        } else {
+            return mqttSubscriber.topic();
+        }
+    }
+    
+    @Override
+    public void destroy() throws Exception {
+        if (mqttClient != null && mqttClient.isConnected()) {
+            for (String topic : topicSubscriberMap.keySet()) {
+                try {
+                    mqttClient.unsubscribe(topic);
+                    log.info("Unsubscribed from topic: {}", topic);
+                } catch (Exception e) {
+                    log.error("Error unsubscribing from topic: {}", topic, e);
+                }
+            }
+        }
+    }
+    
+    /**
+     * 订阅方法信息类
+     */
+    private static class SubscriberMethod {
+        private final Object bean;
+        private final Method method;
+        private final MqttSubscriber mqttSubscriber;
+        
+        public SubscriberMethod(Object bean, Method method, MqttSubscriber mqttSubscriber) {
+            this.bean = bean;
+            this.method = method;
+            this.mqttSubscriber = mqttSubscriber;
+        }
+        
+        public Object getBean() {
+            return bean;
+        }
+        
+        public Method getMethod() {
+            return method;
+        }
+        
+        public MqttSubscriber getMqttSubscriber() {
+            return mqttSubscriber;
+        }
+    }
+} 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/handler/MqttTopicHandler.java

@@ -0,0 +1,18 @@
+package cn.hfln.framework.mqtt.handler;
+
+/**
+ * MQTT主题处理器接口
+ * 定义处理特定主题MQTT消息的方法
+ */
+public interface MqttTopicHandler {
+    
+    /**
+     * 处理MQTT消息
+     * @param topic 消息主题
+     * @param payload 消息内容
+     * @throws Exception 处理异常
+     */
+    void handle(String topic, String payload) throws Exception;
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/interceptor/DefaultMqttMessageInterceptor.java

@@ -28,4 +28,6 @@ public class DefaultMqttMessageInterceptor implements MqttMessageInterceptor {
     public void postReceive(String topic, MqttMessage message) {
 //        log.debug("Post-receive message from topic: {}", topic);
     }
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/interceptor/MqttMessageInterceptor.java

@@ -3,40 +3,65 @@ package cn.hfln.framework.mqtt.interceptor;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
 /**
- * MQTT消息拦截器接口
+ * MQTT娑堟伅鎷︽埅鍣ㄦ帴鍙?
+ * 鐢ㄤ簬鍦ㄦ秷鎭彂閫佸拰鎺ユ敹鍓嶅悗杩涜澶勭悊
  */
 public interface MqttMessageInterceptor {
+    
     /**
-     * 发送消息前拦截
-     *
-     * @param topic   主题
-     * @param message 消息
-     * @return 是否继续发送
+     * 娑堟伅鍙戦€佸墠澶勭悊
+     * 
+     * @param topic 涓婚
+     * @param message 娑堟伅
+     * @return 鏄惁缁х画鍙戦€?
      */
     boolean preSend(String topic, MqttMessage message);
-
+    
     /**
-     * 发送消息后拦截
-     *
-     * @param topic   主题
-     * @param message 消息
+     * 娑堟伅鍙戦€佸悗澶勭悊
+     * 
+     * @param topic 涓婚
+     * @param message 娑堟伅
      */
     void postSend(String topic, MqttMessage message);
-
+    
     /**
-     * 接收消息前拦截
-     *
-     * @param topic   主题
-     * @param message 消息
-     * @return 是否继续处理
+     * 娑堟伅鎺ユ敹鍓嶅鐞?
+     * 
+     * @param topic 涓婚
+     * @param message 娑堟伅
+     * @return 鏄惁缁х画澶勭悊
      */
     boolean preReceive(String topic, MqttMessage message);
-
+    
     /**
-     * 接收消息后拦截
-     *
-     * @param topic   主题
-     * @param message 消息
+     * 娑堟伅鎺ユ敹鍚庡鐞?
+     * 
+     * @param topic 涓婚
+     * @param message 娑堟伅
      */
     void postReceive(String topic, MqttMessage message);
-} 
+    
+    /**
+     * 娑堟伅鍙戝竷鍓嶅鐞?
+     * 
+     * @param topic 涓婚
+     * @param payload 娑堟伅鍐呭
+     * @return 澶勭悊鍚庣殑娑堟伅鍐呭
+     */
+    default String beforePublish(String topic, String payload) {
+        return payload;
+    }
+    
+    /**
+     * 娑堟伅鍙戝竷鍚庡鐞?
+     * 
+     * @param topic 涓婚
+     * @param payload 娑堟伅鍐呭
+     */
+    default void afterPublish(String topic, String payload) {
+        // 榛樿绌哄疄鐜?
+    }
+}
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/listener/DefaultMqttConnectionListener.java

@@ -20,4 +20,6 @@ public class DefaultMqttConnectionListener implements MqttConnectionListener {
     public void onConnectionFailed(Throwable cause) {
         log.error("MQTT client connection failed", cause);
     }
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/listener/DefaultMqttMessageListener.java

@@ -12,4 +12,6 @@ public class DefaultMqttMessageListener implements MqttMessageListener {
         log.info("Received message on topic: {}", topic);
         log.info("Message content: {}", new String(message.getPayload()));
     }
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/listener/MqttConnectionListener.java

@@ -1,5 +1,7 @@
 package cn.hfln.framework.mqtt.listener;
 
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+
 /**
  * MQTT连接状态监听器接口
  */
@@ -22,4 +24,24 @@ public interface MqttConnectionListener {
      * @param cause 失败原因
      */
     void onConnectionFailed(Throwable cause);
-} 
+    
+    /**
+     * 连接丢失回调
+     *
+     * @param cause 丢失原因
+     */
+    default void connectionLost(Throwable cause) {
+        onDisconnected(cause);
+    }
+    
+    /**
+     * 消息发送完成回调
+     *
+     * @param token 消息发送令牌
+     */
+    default void deliveryComplete(IMqttDeliveryToken token) {
+        // 榛樿绌哄疄鐜
+    }
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/listener/MqttMessageListener.java

@@ -13,4 +13,6 @@ public interface MqttMessageListener {
      * @param message 消息
      */
     void onMessage(String topic, MqttMessage message);
-} 
+} 
+
+

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/java/cn/hfln/framework/mqtt/template/MqttTemplate.java

@@ -1,32 +1,121 @@
 package cn.hfln.framework.mqtt.template;
 
+import cn.hfln.framework.mqtt.config.MqttProperties;
+import cn.hfln.framework.mqtt.connection.MqttReconnectManager;
+import cn.hfln.framework.mqtt.converter.JsonMessageConverter;
 import cn.hfln.framework.mqtt.converter.MessageConverter;
+import cn.hfln.framework.mqtt.exception.MqttExceptionHandler;
 import cn.hfln.framework.mqtt.interceptor.MqttMessageInterceptor;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
-import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
 
-import java.nio.charset.StandardCharsets;
+import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
 
+/**
+ * MQTT消息模板
+ * 提供发送MQTT消息的便捷方法
+ */
 @Slf4j
 @Component
-public class MqttTemplate {
-    private final MqttClient mqttClient;
-    private final ObjectMapper objectMapper;
+public class MqttTemplate implements DisposableBean {
+    private final MqttPahoMessageHandler mqttOutbound;
+    private final JsonMessageConverter jsonConverter;
     private final List<MqttMessageInterceptor> interceptors;
+    private final MqttClient mqttClient;
     private final MessageConverter messageConverter;
+    private final MqttProperties mqttProperties;
+    private final MqttReconnectManager reconnectManager;
+    
+    // 用于重试发送消息的线程池
+    private final ScheduledExecutorService retryExecutor;
+    
+    // 消息发送统计
+    private final AtomicInteger totalMessagesSent = new AtomicInteger(0);
+    private final AtomicInteger failedMessages = new AtomicInteger(0);
+    private final AtomicInteger retriedMessages = new AtomicInteger(0);
+
+    @Autowired(required = false)
+    public MqttTemplate(MqttPahoMessageHandler mqttOutbound, JsonMessageConverter jsonConverter) {
+        this.mqttOutbound = mqttOutbound;
+        this.jsonConverter = jsonConverter;
+        this.interceptors = Collections.emptyList();
+        this.mqttClient = null;
+        this.messageConverter = null;
+        this.mqttProperties = null;
+        this.reconnectManager = null;
+        this.retryExecutor = null;
+    }
 
-    @Autowired
-    public MqttTemplate(MqttClient mqttClient, List<MqttMessageInterceptor> interceptors, MessageConverter messageConverter) {
+    @Autowired(required = false)
+    public MqttTemplate(MqttPahoMessageHandler mqttOutbound, JsonMessageConverter jsonConverter, 
+                      List<MqttMessageInterceptor> interceptors) {
+        this.mqttOutbound = mqttOutbound;
+        this.jsonConverter = jsonConverter;
+        this.interceptors = interceptors != null ? interceptors : Collections.emptyList();
+        this.mqttClient = null;
+        this.messageConverter = null;
+        this.mqttProperties = null;
+        this.reconnectManager = null;
+        this.retryExecutor = null;
+    }
+    
+    @Autowired(required = false)
+    public MqttTemplate(MqttClient mqttClient, List<MqttMessageInterceptor> interceptors,
+                     MessageConverter messageConverter) {
+        this.mqttClient = mqttClient;
+        this.messageConverter = messageConverter;
+        this.interceptors = interceptors != null ? interceptors : Collections.emptyList();
+        this.mqttOutbound = null;
+        this.jsonConverter = (messageConverter instanceof JsonMessageConverter) ? 
+                            (JsonMessageConverter) messageConverter : null;
+        this.mqttProperties = null;
+        this.reconnectManager = null;
+        this.retryExecutor = null;
+    }
+    
+    @Autowired(required = false)
+    public MqttTemplate(MqttClient mqttClient, List<MqttMessageInterceptor> interceptors,
+                     MessageConverter messageConverter, MqttProperties mqttProperties,
+                     MqttReconnectManager reconnectManager) {
         this.mqttClient = mqttClient;
-        this.objectMapper = new ObjectMapper();
-        this.interceptors = interceptors;
         this.messageConverter = messageConverter;
+        this.interceptors = interceptors != null ? interceptors : Collections.emptyList();
+        this.mqttOutbound = null;
+        this.jsonConverter = (messageConverter instanceof JsonMessageConverter) ? 
+                            (JsonMessageConverter) messageConverter : null;
+        this.mqttProperties = mqttProperties;
+        this.reconnectManager = reconnectManager;
+        
+        // 创建重试线程池
+        ThreadFactory threadFactory = r -> {
+            Thread t = new Thread(r, "mqtt-retry-thread");
+            t.setDaemon(true);
+            return t;
+        };
+        
+        this.retryExecutor = Executors.newScheduledThreadPool(
+            Runtime.getRuntime().availableProcessors(), 
+            threadFactory
+        );
+        
+        log.info("MqttTemplate initialized with {} interceptors", this.interceptors.size());
+        
+        // 添加JVM关闭钩子，确保线程池正确关闭
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            shutdown();
+        }));
     }
 
     /**
@@ -37,7 +126,7 @@ public class MqttTemplate {
      * @return CompletableFuture<Void>
      */
     public CompletableFuture<Void> send(String topic, String payload) {
-        return send(topic, payload.getBytes(StandardCharsets.UTF_8), 0, false);
+        return send(topic, payload, mqttProperties != null ? mqttProperties.getDefaultQos() : 0, false);
     }
 
     /**
@@ -50,88 +139,270 @@ public class MqttTemplate {
      * @return CompletableFuture<Void>
      */
     public CompletableFuture<Void> send(String topic, String payload, int qos, boolean retain) {
-        return send(topic, payload.getBytes(StandardCharsets.UTF_8), qos, retain);
-    }
-
-    /**
-     * 发送字节数组消息
-     *
-     * @param topic   主题
-     * @param payload 消息内容
-     * @return CompletableFuture<Void>
-     */
-    public CompletableFuture<Void> send(String topic, byte[] payload) {
-        return send(topic, payload, 0, false);
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        totalMessagesSent.incrementAndGet();
+        
+        // 应用拦截器
+        String processedPayload = payload;
+        for (MqttMessageInterceptor interceptor : interceptors) {
+            try {
+                processedPayload = interceptor.beforePublish(topic, processedPayload);
+            } catch (Exception e) {
+                log.error("Error in message interceptor", e);
+                failedMessages.incrementAndGet();
+                future.completeExceptionally(e);
+                return future;
+            }
+        }
+        
+        try {
+            final String finalPayload = processedPayload;
+            if (mqttOutbound != null) {
+                Message<String> message = MessageBuilder
+                        .withPayload(finalPayload)
+                        .setHeader(MqttHeaders.TOPIC, topic)
+                        .setHeader(MqttHeaders.QOS, qos)
+                        .setHeader(MqttHeaders.RETAINED, retain)
+                        .build();
+                
+                mqttOutbound.handleMessage(message);
+                
+                // 通知拦截器
+                for (MqttMessageInterceptor interceptor : interceptors) {
+                    try {
+                        interceptor.afterPublish(topic, finalPayload);
+                    } catch (Exception e) {
+                        log.error("Error in message interceptor after publish", e);
+                    }
+                }
+                
+                future.complete(null);
+                log.debug("Sent message to topic: {}", topic);
+            } else if (mqttClient != null) {
+                publishWithRetry(topic, finalPayload.getBytes(), qos, retain, future, 0);
+                return future;
+            } else {
+                throw new IllegalStateException("No MQTT client or message handler available");
+            }
+        } catch (Exception e) {
+            failedMessages.incrementAndGet();
+            
+            // 使用异常处理器处理异常
+            if (e instanceof MqttException) {
+                final String finalPayload = processedPayload;
+                MqttExceptionHandler.handleException((MqttException) e, ex -> {
+                    if (mqttProperties != null && mqttProperties.isRetryOnPublishFailure() && 
+                        MqttExceptionHandler.isRecoverableByReconnect((MqttException) e)) {
+                        if (retryExecutor != null && !retryExecutor.isShutdown()) {
+                            log.info("Scheduling retry for message to topic: {}", topic);
+                            retryExecutor.schedule(() -> {
+                                retriedMessages.incrementAndGet();
+                                send(topic, finalPayload, qos, retain).thenAccept(v -> future.complete(null))
+                                    .exceptionally(t -> {
+                                        future.completeExceptionally(t);
+                                        return null;
+                                    });
+                            }, mqttProperties.getPublishRetryInterval(), TimeUnit.MILLISECONDS);
+                        } else {
+                            future.completeExceptionally(e);
+                        }
+                    } else {
+                        future.completeExceptionally(e);
+                    }
+                });
+            } else {
+                future.completeExceptionally(e);
+                log.error("Failed to send message to topic: {}", topic, e);
+            }
+        }
+        return future;
     }
-
+    
     /**
-     * 发送字节数组消息（指定QoS和保留标志）
-     *
-     * @param topic   主题
+     * 带重试机制的消息发布
+     * 
+     * @param topic 主题
      * @param payload 消息内容
-     * @param qos     QoS级别
-     * @param retain  是否保留
-     * @return CompletableFuture<Void>
+     * @param qos QoS级别
+     * @param retain 是否保留
+     * @param future 完成回调
+     * @param retryCount 当前重试次数
      */
-    public CompletableFuture<Void> send(String topic, byte[] payload, int qos, boolean retain) {
-        CompletableFuture<Void> future = new CompletableFuture<>();
+    private void publishWithRetry(String topic, byte[] payload, int qos, boolean retain, 
+                                CompletableFuture<Void> future, int retryCount) {
         try {
             MqttMessage message = new MqttMessage(payload);
             message.setQos(qos);
             message.setRetained(retain);
-
-            // 执行发送前拦截
-            boolean shouldSend = interceptors.stream()
-                    .allMatch(interceptor -> interceptor.preSend(topic, message));
-
-            if (shouldSend) {
-                mqttClient.publish(topic, message);
-                // 执行发送后拦截
-                interceptors.forEach(interceptor -> interceptor.postSend(topic, message));
-                future.complete(null);
-            } else {
-                future.completeExceptionally(new RuntimeException("Message sending was intercepted"));
+            
+            if (!mqttClient.isConnected()) {
+                if (mqttProperties != null && mqttProperties.isAutomaticReconnect() && reconnectManager != null) {
+                    log.warn("MQTT client not connected, triggering reconnect before sending message");
+                    reconnectManager.triggerReconnect();
+                    
+                    // 如果启用了重试，则安排重试
+                    if (shouldRetry(retryCount)) {
+                        retriedMessages.incrementAndGet();
+                        scheduleRetry(topic, payload, qos, retain, future, retryCount + 1);
+                        return;
+                    } else {
+                        failedMessages.incrementAndGet();
+                        future.completeExceptionally(new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED));
+                        return;
+                    }
+                } else {
+                    failedMessages.incrementAndGet();
+                    future.completeExceptionally(new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED));
+                    return;
+                }
+            }
+            
+            // 应用拦截器
+            for (MqttMessageInterceptor interceptor : interceptors) {
+                if (!interceptor.preSend(topic, message)) {
+                    log.warn("Message sending cancelled by interceptor for topic: {}", topic);
+                    failedMessages.incrementAndGet();
+                    future.completeExceptionally(new IllegalStateException("Message sending cancelled by interceptor"));
+                    return;
+                }
+            }
+            
+            mqttClient.publish(topic, message);
+            future.complete(null);
+            log.debug("Successfully published message to topic: {}", topic);
+            
+            // 通知拦截器
+            for (MqttMessageInterceptor interceptor : interceptors) {
+                try {
+                    interceptor.postSend(topic, message);
+                    interceptor.afterPublish(topic, new String(payload));
+                } catch (Exception e) {
+                    log.error("Error in message interceptor after publish", e);
+                }
             }
         } catch (MqttException e) {
-            future.completeExceptionally(e);
+            // 处理异常
+            String detailedError = MqttExceptionHandler.getDetailedErrorMessage(e);
+            log.error("Failed to publish message to topic {}: {}", topic, detailedError);
+            
+            // 如果是可恢复错误并且启用了重试，则安排重试
+            if (MqttExceptionHandler.isRecoverableByReconnect(e) && shouldRetry(retryCount)) {
+                log.info("Will retry publishing message to topic {} (attempt {}/{})", 
+                        topic, retryCount + 1, mqttProperties.getMaxPublishRetryAttempts());
+                retriedMessages.incrementAndGet();
+                scheduleRetry(topic, payload, qos, retain, future, retryCount + 1);
+            } else {
+                failedMessages.incrementAndGet();
+                future.completeExceptionally(e);
+                
+                // 提供解决方案建议
+                String recommendation = MqttExceptionHandler.getRecommendedAction(e);
+                log.info("Recommended action: {}", recommendation);
+            }
         }
-        return future;
+    }
+    
+    /**
+     * 判断是否应该重试发送消息
+     * 
+     * @param retryCount 当前重试次数
+     * @return 是否应该重试
+     */
+    private boolean shouldRetry(int retryCount) {
+        return mqttProperties != null && 
+               mqttProperties.isRetryOnPublishFailure() && 
+               retryCount < mqttProperties.getMaxPublishRetryAttempts();
+    }
+    
+    /**
+     * 安排重试发送消息
+     * 
+     * @param topic 主题
+     * @param payload 消息内容
+     * @param qos QoS级别
+     * @param retain 是否保留
+     * @param future 完成回调
+     * @param retryCount 当前重试次数
+     */
+    private void scheduleRetry(String topic, byte[] payload, int qos, boolean retain, 
+                             CompletableFuture<Void> future, int retryCount) {
+        if (retryExecutor != null && !retryExecutor.isShutdown()) {
+            long delay = calculateRetryDelay(retryCount);
+            log.debug("Scheduling retry {} for topic {} in {} ms", retryCount, topic, delay);
+            
+            retryExecutor.schedule(
+                () -> publishWithRetry(topic, payload, qos, retain, future, retryCount),
+                delay,
+                TimeUnit.MILLISECONDS
+            );
+        } else {
+            future.completeExceptionally(new IllegalStateException("Retry executor is not available"));
+        }
+    }
+    
+    /**
+     * 计算重试延迟时间，使用指数退避算法
+     * 
+     * @param retryCount 当前重试次数
+     * @return 延迟时间（毫秒）
+     */
+    private long calculateRetryDelay(int retryCount) {
+        if (mqttProperties == null) {
+            return 1000; // 默认1秒
+        }
+        
+        long baseDelay = mqttProperties.getPublishRetryInterval();
+        double factor = Math.pow(mqttProperties.getReconnectBackoffMultiplier(), retryCount - 1);
+        long delay = (long)(baseDelay * factor);
+        
+        // 确保不超过最大重试间隔
+        return Math.min(delay, mqttProperties.getMaxReconnectInterval());
     }
 
     /**
      * 发送JSON对象消息
      *
      * @param topic 主题
-     * @param obj   对象
+     * @param payload 消息内容对象
      * @return CompletableFuture<Void>
      */
-    public CompletableFuture<Void> sendJson(String topic, Object obj) {
-        try {
-            String json = objectMapper.writeValueAsString(obj);
-            return send(topic, json);
-        } catch (JsonProcessingException e) {
-            CompletableFuture<Void> future = new CompletableFuture<>();
-            future.completeExceptionally(e);
-            return future;
-        }
+    public CompletableFuture<Void> sendJson(String topic, Object payload) {
+        return sendJson(topic, payload, mqttProperties != null ? mqttProperties.getDefaultQos() : 0, false);
     }
 
     /**
      * 发送JSON对象消息（指定QoS和保留标志）
      *
      * @param topic  主题
-     * @param obj    对象
+     * @param payload 消息内容对象
      * @param qos    QoS级别
      * @param retain 是否保留
      * @return CompletableFuture<Void>
      */
-    public CompletableFuture<Void> sendJson(String topic, Object obj, int qos, boolean retain) {
+    public CompletableFuture<Void> sendJson(String topic, Object payload, int qos, boolean retain) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
         try {
-            String json = objectMapper.writeValueAsString(obj);
-            return send(topic, json, qos, retain);
-        } catch (JsonProcessingException e) {
-            CompletableFuture<Void> future = new CompletableFuture<>();
+            if (jsonConverter != null) {
+                String json = jsonConverter.toJson(payload);
+                return send(topic, json, qos, retain);
+            } else if (messageConverter != null) {
+                MqttMessage message = messageConverter.toMessage(payload);
+                message.setQos(qos);
+                message.setRetained(retain);
+                
+                if (mqttClient != null) {
+                    publishWithRetry(topic, message.getPayload(), qos, retain, future, 0);
+                    return future;
+                } else {
+                    throw new IllegalStateException("No MQTT client available");
+                }
+            } else {
+                throw new IllegalStateException("No message converter available");
+            }
+        } catch (Exception e) {
+            failedMessages.incrementAndGet();
             future.completeExceptionally(e);
+            log.error("Failed to convert object to JSON: {}", payload, e);
             return future;
         }
     }
@@ -141,10 +412,10 @@ public class MqttTemplate {
      *
      * @param topic   主题
      * @param payload 消息内容
-     * @throws MqttException 发送异常
+     * @throws Exception 发送异常
      */
-    public void sendSync(String topic, String payload) throws MqttException {
-        sendSync(topic, payload.getBytes(StandardCharsets.UTF_8), 0, false);
+    public void sendSync(String topic, String payload) throws Exception {
+        sendSync(topic, payload, mqttProperties != null ? mqttProperties.getDefaultQos() : 0, false);
     }
 
     /**
@@ -154,23 +425,25 @@ public class MqttTemplate {
      * @param payload 消息内容
      * @param qos     QoS级别
      * @param retain  是否保留
-     * @throws MqttException 发送异常
+     * @throws Exception 发送异常
      */
-    public void sendSync(String topic, byte[] payload, int qos, boolean retain) throws MqttException {
-        MqttMessage message = new MqttMessage(payload);
-        message.setQos(qos);
-        message.setRetained(retain);
-
-        // 执行发送前拦截
-        boolean shouldSend = interceptors.stream()
-                .allMatch(interceptor -> interceptor.preSend(topic, message));
-
-        if (shouldSend) {
-            mqttClient.publish(topic, message);
-            // 执行发送后拦截
-            interceptors.forEach(interceptor -> interceptor.postSend(topic, message));
-        } else {
-            throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION);
+    public void sendSync(String topic, String payload, int qos, boolean retain) throws Exception {
+        try {
+            send(topic, payload, qos, retain).get(
+                mqttProperties != null ? mqttProperties.getCompletionTimeout() : 5000, 
+                TimeUnit.MILLISECONDS
+            );
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION, e);
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof MqttException) {
+                throw (MqttException) e.getCause();
+            } else {
+                throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION, e.getCause());
+            }
+        } catch (TimeoutException e) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_TIMEOUT, e);
         }
     }
 
@@ -178,33 +451,89 @@ public class MqttTemplate {
      * 同步发送JSON对象消息
      *
      * @param topic 主题
-     * @param obj   对象
-     * @throws MqttException 发送异常
+     * @param payload 消息内容对象
+     * @throws Exception 发送异常
      */
-    public void sendJsonSync(String topic, Object obj) throws MqttException {
-        try {
-            String json = objectMapper.writeValueAsString(obj);
-            sendSync(topic, json);
-        } catch (JsonProcessingException e) {
-            throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION, e);
-        }
+    public void sendJsonSync(String topic, Object payload) throws Exception {
+        sendJsonSync(topic, payload, mqttProperties != null ? mqttProperties.getDefaultQos() : 0, false);
     }
 
     /**
      * 同步发送JSON对象消息（指定QoS和保留标志）
      *
      * @param topic  主题
-     * @param obj    对象
+     * @param payload 消息内容对象
      * @param qos    QoS级别
      * @param retain 是否保留
-     * @throws MqttException 发送异常
+     * @throws Exception 发送异常
      */
-    public void sendJsonSync(String topic, Object obj, int qos, boolean retain) throws MqttException {
+    public void sendJsonSync(String topic, Object payload, int qos, boolean retain) throws Exception {
         try {
-            String json = objectMapper.writeValueAsString(obj);
-            sendSync(topic, json.getBytes(StandardCharsets.UTF_8), qos, retain);
-        } catch (JsonProcessingException e) {
+            sendJson(topic, payload, qos, retain).get(
+                mqttProperties != null ? mqttProperties.getCompletionTimeout() : 5000, 
+                TimeUnit.MILLISECONDS
+            );
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION, e);
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof MqttException) {
+                throw (MqttException) e.getCause();
+            } else {
+                throw new MqttException(MqttException.REASON_CODE_CLIENT_EXCEPTION, e.getCause());
+            }
+        } catch (TimeoutException e) {
+            throw new MqttException(MqttException.REASON_CODE_CLIENT_TIMEOUT, e);
         }
     }
+    
+    /**
+     * 获取消息发送统计信息
+     * 
+     * @return 统计信息
+     */
+    public String getStatistics() {
+        int total = totalMessagesSent.get();
+        int failed = failedMessages.get();
+        int retried = retriedMessages.get();
+        int success = total - failed;
+        double successRate = total > 0 ? (double)success / total * 100 : 0;
+        
+        return String.format(
+            "Messages - Total: %d, Success: %d (%.2f%%), Failed: %d, Retried: %d",
+            total, success, successRate, failed, retried
+        );
+    }
+    
+    /**
+     * 重置统计信息
+     */
+    public void resetStatistics() {
+        totalMessagesSent.set(0);
+        failedMessages.set(0);
+        retriedMessages.set(0);
+    }
+    
+    /**
+     * 关闭资源
+     */
+    private void shutdown() {
+        if (retryExecutor != null && !retryExecutor.isShutdown()) {
+            log.info("Shutting down MQTT template retry executor");
+            retryExecutor.shutdown();
+            try {
+                if (!retryExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+                    retryExecutor.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                retryExecutor.shutdownNow();
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+    
+    @Override
+    public void destroy() throws Exception {
+        shutdown();
+    }
 } 

hfln-framework-design-starter/mqtt-spring-boot-starter/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+cn.hfln.framework.mqtt.config.MqttAutoConfiguration 

pom.xml

@@ -45,6 +45,13 @@
         <fastjson.version>2.0.41</fastjson.version>
         <transmittable.version>2.14.2</transmittable.version>
         <jjwt.version>0.9.1</jjwt.version>
+
+        <eclipse-mqttv.version>1.2.5</eclipse-mqttv.version>
+        <spring-integration.version>5.5.14</spring-integration.version>
+        <junit-jupiter.version>5.6.3</junit-jupiter.version>
+        <logback.version>1.2.11</logback.version>
+        <mockito.version>4.5.1</mockito.version>
+        <byte-buddy.version>1.12.10</byte-buddy.version>
     </properties>
 
     <modules>
@@ -203,6 +210,74 @@
                 <version>${knife4j.version}</version>
             </dependency>
 
+
+            <dependency>
+                <groupId>org.eclipse.paho</groupId>
+                <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+                <version>${eclipse-mqttv.version}</version>
+            </dependency>
+
+            <!-- JUnit Jupiter -->
+            <dependency>
+                <groupId>org.junit.jupiter</groupId>
+                <artifactId>junit-jupiter-api</artifactId>
+                <version>${junit-jupiter.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.junit.jupiter</groupId>
+                <artifactId>junit-jupiter-engine</artifactId>
+                <version>${junit-jupiter.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.junit.jupiter</groupId>
+                <artifactId>junit-jupiter</artifactId>
+                <version>${junit-jupiter.version}</version>
+                <optional>true</optional>
+                <scope>test</scope>
+            </dependency>
+
+            <!-- Logging -->
+            <dependency>
+                <groupId>ch.qos.logback</groupId>
+                <artifactId>logback-classic</artifactId>
+                <version>${logback.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>ch.qos.logback</groupId>
+                <artifactId>logback-core</artifactId>
+                <version>${logback.version}</version>
+                <scope>test</scope>
+            </dependency>
+
+            <dependency>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-core</artifactId>
+                <version>${mockito.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-junit-jupiter</artifactId>
+                <version>${mockito.version}</version>
+                <scope>test</scope>
+            </dependency>
+
+            <dependency>
+                <groupId>net.bytebuddy</groupId>
+                <artifactId>byte-buddy</artifactId>
+                <version>${byte-buddy.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>net.bytebuddy</groupId>
+                <artifactId>byte-buddy-agent</artifactId>
+                <version>${byte-buddy.version}</version>
+                <scope>test</scope>
+            </dependency>
+
             <dependency>
                 <groupId>cn.hfln.framework</groupId>
                 <artifactId>hfln-framework-common</artifactId>