MQTT方法优化

@@ -253,6 +253,7 @@ public class MqttUtils {
             // 连接到服务器
             // 连接到服务器
             log.info("正在连接到: " + brokerUrl);
             log.info("正在连接到: " + brokerUrl);
             mqttClient.connect(connectOptions);
             mqttClient.connect(connectOptions);
+            isConnected = mqttClient.isConnected();
 
 
         } catch (MqttException e) {
         } catch (MqttException e) {
             isConnected = false;
             isConnected = false;
@@ -274,21 +275,54 @@ public class MqttUtils {
      * @throws MqttException 发布异常
      * @throws MqttException 发布异常
      */
      */
     public void publish(String topic, String message, int qos, boolean retained) throws MqttException {
     public void publish(String topic, String message, int qos, boolean retained) throws MqttException {
-        if (!isConnected) {
+        if (mqttClient == null) {
             throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
             throw new MqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
         }
         }
 
 
-        try {
-            MqttMessage mqttMessage = new MqttMessage(message.getBytes(StandardCharsets.UTF_8));
-            mqttMessage.setQos(qos);
-            mqttMessage.setRetained(retained);
+        int attempts = 0;
+        while (true) {
+            if (!mqttClient.isConnected()) {
+                try {
+                    if (connectOptions != null) {
+                        mqttClient.connect(connectOptions);
+                        isConnected = mqttClient.isConnected();
+                    }
+                } catch (MqttException ce) {
+                    attempts++;
+                    if (attempts >= 2) {
+                        log.error("重连失败: " + ce.getMessage());
+                        throw ce;
+                    }
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                    }
+                    continue;
+                }
+            }
 
 
-            mqttClient.publish(topic, mqttMessage);
-            log.info("发布成功 - Topic: " + topic + ", QoS: " + qos +
-                    ", Retained: " + retained + ", Message: " + message);
-        } catch (MqttException e) {
-            log.error("发布失败: " + e.getMessage());
-            throw e;
+            try {
+                MqttMessage mqttMessage = new MqttMessage(message.getBytes(StandardCharsets.UTF_8));
+                mqttMessage.setQos(qos);
+                mqttMessage.setRetained(retained);
+
+                mqttClient.publish(topic, mqttMessage);
+                log.info("发布成功 - Topic: " + topic + ", QoS: " + qos +
+                        ", Retained: " + retained + ", Message: " + message);
+                return;
+            } catch (MqttException e) {
+                attempts++;
+                if (attempts >= 2) {
+                    log.error("发布失败: " + e.getMessage());
+                    throw e;
+                }
+                try {
+                    Thread.sleep(100);
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                }
+            }
         }
         }
     }
     }
 
 
@@ -470,7 +504,8 @@ public class MqttUtils {
 
 
         config.setConnectionTimeout(15)
         config.setConnectionTimeout(15)
                 .setKeepAliveInterval(30)
                 .setKeepAliveInterval(30)
-                .setQos(1);
+                .setQos(1)
+                .setMaxInflight(50);
 
 
         try {
         try {
             mqttUtils.connect(config);
             mqttUtils.connect(config);
@@ -490,19 +525,28 @@ public class MqttUtils {
                 .setPassword(password)
                 .setPassword(password)
                 .setConnectionTimeout(15)
                 .setConnectionTimeout(15)
                 .setKeepAliveInterval(30)
                 .setKeepAliveInterval(30)
-                .setQos(1);
+                .setQos(1)
+                .setMaxInflight(50);
 
 
         try {
         try {
-            // 连接
             mqttUtils.connect(config);
             mqttUtils.connect(config);
 
 
-            // 发布消息
             mqttUtils.publish(topic, message);
             mqttUtils.publish(topic, message);
 
 
         } catch (MqttException e) {
         } catch (MqttException e) {
-            e.printStackTrace();
+            log.error("首次发布失败: " + e.getMessage());
+            try {
+                mqttUtils.disconnect();
+            } catch (Exception ignore) {
+            }
+            try {
+                mqttUtils.connect(config);
+                mqttUtils.publish(topic, message);
+            } catch (MqttException ee) {
+                log.error("重试发布失败: " + ee.getMessage());
+                throw ee;
+            }
         } finally {
         } finally {
-            // 断开连接
             mqttUtils.disconnect();
             mqttUtils.disconnect();
         }
         }
     }
     }