/*华软MQTT服务器信息改为由@ConfigurationProperties来读取服务器配置信息*/

inspect-metadata/src/main/java/com/inspect/metadata/mqtt/MQTTUtils.java

@@ -3,7 +3,6 @@ package com.inspect.metadata.mqtt;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
@@ -12,50 +11,44 @@ import javax.annotation.PreDestroy;
 @Slf4j
 @Service
 public class MQTTUtils {
-    @Value("${huaRuan.mqtt.ip:localhost}")
-    protected String ip;
-    @Value("${huaRuan.mqtt.port:1883}")
-    protected String port;
-    @Value("${huaRuan.mqtt.clientId:test}")
-    protected String clientId;
-    @Value("${huaRuan.mqtt.user:admin}")
-    protected String user;
-    @Value("${huaRuan.mqtt.pwd:123456}")
-    protected String pwd;
-    @Value("${huaRuan.mqtt.topicFilters:test}")
-    protected String topicFilters;
-    @Value("${mqtt.qos:1}")
-    protected int[] qos; // QoS 数组
-
-    private MqttClient client = null;
+
+    private final MqttProperties properties;
+
+    private MqttClient client;
     private Long id = 0L;
 
+    public MQTTUtils(MqttProperties properties) {
+        this.properties = properties;
+    }
+
     public Long getId() {
         return this.id;
     }
 
     @PostConstruct
-    public void mqttUtils() {
-        log.info("mqttUtil start!!!");
+    public void init() {
+        log.info("MQTT Utils initializing...");
         connectWithRetry();
     }
 
     private void connectWithRetry() {
         MemoryPersistence persistence = new MemoryPersistence();
         MqttConnectOptions connOpts = new MqttConnectOptions();
-        connOpts.setUserName(user);
-        connOpts.setPassword(pwd.toCharArray());
+        connOpts.setUserName(properties.getUser());
+        connOpts.setPassword(properties.getPwd().toCharArray());
         connOpts.setCleanSession(true);
 
-        String broker = String.format("tcp://%s:%s", ip, port);
+        String broker = String.format("tcp://%s:%d", properties.getIp(), properties.getPort());
 
         while (true) {
             try {
-                client = new MqttClient(broker, clientId, persistence);
+                client = new MqttClient(broker, properties.getClientId(), persistence);
                 client.setCallback(new MqttCallback() {
                     @Override
                     public void connectionLost(Throwable cause) {
                         log.error("Connection lost: {}", cause.getMessage());
+                        // 自动重连
+                        connectWithRetry();
                     }
 
                     @Override
@@ -65,7 +58,7 @@ public class MQTTUtils {
 
                     @Override
                     public void deliveryComplete(IMqttDeliveryToken token) {
-
+                        log.info("Delivery complete: {}", token.getMessageId());
                     }
                 });
 
@@ -73,15 +66,15 @@ public class MQTTUtils {
                 client.connect(connOpts);
                 log.info("Connected to MQTT broker");
 
-                subscribe(new String[]{topicFilters}, qos);
+                subscribe(properties.getTopicFilters(), properties.getQos());
                 break;
             } catch (MqttException e) {
                 log.error("Failed to connect to MQTT broker. Reason: {}", e.getMessage());
                 try {
                     log.info("Retrying connection in 30 seconds...");
-                    Thread.sleep(30000); // Wait for 30 seconds before retrying
+                    Thread.sleep(30000);
                 } catch (InterruptedException ie) {
-                    Thread.currentThread().interrupt(); // Restore interrupted status
+                    Thread.currentThread().interrupt();
                     log.error("Connection retry interrupted", ie);
                 }
             }
@@ -93,19 +86,21 @@ public class MQTTUtils {
             log.info("Subscribing to topicFilters: {}, qos: {}", topicFilters, qos);
             client.subscribe(topicFilters, qos);
         } catch (MqttException e) {
-            log.error("Failed to subscribe. Reason: {}", e.getMessage());
-            e.printStackTrace();
+            log.error("Failed to subscribe. Reason: {}", e.getMessage(), e);
         }
     }
 
     public void disconnect() {
         try {
             log.info("Disconnecting from MQTT broker");
-            client.disconnect();
-            client.close();
+            if (client != null && client.isConnected()) {
+                client.disconnect();
+            }
+            if (client != null) {
+                client.close();
+            }
         } catch (MqttException e) {
-            log.error("Failed to disconnect. Reason: {}", e.getMessage());
-            e.printStackTrace();
+            log.error("Failed to disconnect. Reason: {}", e.getMessage(), e);
         }
     }
 
@@ -117,8 +112,7 @@ public class MQTTUtils {
             client.publish(pubTopic, message);
             return true;
         } catch (MqttException e) {
-            log.error("Failed to publish message. Reason: {}", e.getMessage());
-            e.printStackTrace();
+            log.error("Failed to publish message. Reason: {}", e.getMessage(), e);
             return false;
         }
     }
@@ -131,4 +125,4 @@ public class MQTTUtils {
     public MqttClient getClient() {
         return client;
     }
-}
+}

inspect-metadata/src/main/java/com/inspect/metadata/mqtt/MqttProperties.java

@@ -0,0 +1,31 @@
+package com.inspect.metadata.mqtt;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+@ConfigurationProperties(prefix = "hua-ruan.mqtt")
+public class MqttProperties {
+    /** MQTT Broker IP */
+    private String ip = "localhost";
+
+    /** MQTT Broker Port */
+    private int port = 1883;
+
+    /** 客户端 ID */
+    private String clientId = "test";
+
+    /** 用户名 */
+    private String user = "admin";
+
+    /** 密码 */
+    private String pwd = "123456";
+
+    /** 订阅的 topic 列表 */
+    private String[] topicFilters = new String[0];
+
+    /** QoS 数组 */
+    private int[] qos = new int[]{1};
+}