mqtt方法优化

@@ -140,7 +140,7 @@ public class CzDeviceReportService {
 
 
             try {
             try {
                 log.debug("开始MQTT发布: deviceId={}", deviceId);
                 log.debug("开始MQTT发布: deviceId={}", deviceId);
-                MqttUtils.quickPublish(broker, topic, username, password, clientId, JSON.toJSONString(deviceMap));
+                MqttUtils.publish(broker, topic, username, password, clientId, JSON.toJSONString(deviceMap));
                 successCount++;
                 successCount++;
 
 
                 // 每10个设备记录一次进度
                 // 每10个设备记录一次进度
@@ -274,4 +274,4 @@ public class CzDeviceReportService {
         log.info("数据库操作完成，返回{}条记录", resultList.size());
         log.info("数据库操作完成，返回{}条记录", resultList.size());
         return resultList;
         return resultList;
     }
     }
-}
+}

 package com.iot.scheduler.utils;
 package com.iot.scheduler.utils;
 
 
+import io.micrometer.common.util.StringUtils;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 
 
@@ -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();
+                }
+            }
         }
         }
     }
     }
 
 
@@ -456,7 +490,7 @@ public class MqttUtils {
                                     String password, String clientId, String message) throws Exception {
                                     String password, String clientId, String message) throws Exception {
         MqttUtils mqttUtils = new MqttUtils();
         MqttUtils mqttUtils = new MqttUtils();
 
 
-        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker);
+        MqttConfig config = new MqttConfig(broker);
 
 
         if (StringUtils.isNotBlank(username)) {
         if (StringUtils.isNotBlank(username)) {
             config.setUsername(username);
             config.setUsername(username);
@@ -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);
@@ -481,6 +516,41 @@ public class MqttUtils {
         }
         }
     }
     }
 
 
+    public static void publish(String broker, String topic, String username,
+                               String password, String clientId, String message) throws Exception {
+        MqttUtils mqttUtils = new MqttUtils();
+        MqttUtils.MqttConfig config = new MqttUtils.MqttConfig(broker)
+                .setClientId(clientId)
+                .setUsername(username)
+                .setPassword(password)
+                .setConnectionTimeout(15)
+                .setKeepAliveInterval(30)
+                .setQos(1)
+                .setMaxInflight(50);
+
+        try {
+            mqttUtils.connect(config);
+
+            mqttUtils.publish(topic, message);
+
+        } catch (MqttException e) {
+            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 {
+            mqttUtils.disconnect();
+        }
+    }
+
 
 
     public static void main(String[] args) {
     public static void main(String[] args) {
 
 
@@ -520,4 +590,4 @@ public class MqttUtils {
 //            mqttUtils.disconnect();
 //            mqttUtils.disconnect();
 //        }
 //        }
     }
     }
-}
+}