fix: mqtt连接

server-common/src/main/java/com/makeit/module/iot/util/HeaderUtils.java

@@ -70,7 +70,7 @@ public class HeaderUtils {
         headers.put("X-Client-Id", clientId);
         headers.put("X-Timestamp", xTimestamp);
 
-        log.info("请求头信息：{}",Hex.encodeHexString(digest.digest()) +"|"+clientId+"|"+ xTimestamp);
+        // log.info("请求头信息：{}",Hex.encodeHexString(digest.digest()) +"|"+clientId+"|"+ xTimestamp);
         return headers;
     }
 

server-common/src/main/java/com/makeit/module/iot/vo/HeaderInfo.java

 package com.makeit.module.iot.vo;
 
+import com.google.common.collect.Lists;
 import lombok.Data;
 
 import java.util.List;
@@ -13,7 +14,7 @@ public class HeaderInfo {
     private String _uid;
     private String creatorId;
     private String traceparent;
-    private List<Bind> bindings;
+    private List<Bind> bindings = Lists.newArrayList();
 
 
     @Data

server-web/src/main/java/com/makeit/mqtt/MqttPushClient.java

@@ -47,6 +47,7 @@ public class MqttPushClient {
             options.setPassword(password.toCharArray());
             options.setConnectionTimeout(timeout);
             options.setKeepAliveInterval(keepalive);
+            options.setAutomaticReconnect(true);
             MqttPushClient.setMqttClient(client);
             if (!client.isConnected()) {
                 try {

server-web/src/main/java/com/makeit/mqtt/PushCallback.java

@@ -98,41 +98,46 @@ public class PushCallback implements MqttCallbackExtended {
 
     @Override
     public void connectionLost(Throwable cause) {
-        logger.info("连接断开，可以重连");
-        if (client == null || !client.isConnected()) {
-            mqttConfig.getMqttPushClient();
+        logger.info("连接断开，可以重连", cause);
+        int times = 1;
+        while (client == null || !client.isConnected()) {
+            if (times > 5) {
+                logger.info("已达最大次数");
+                break;
+            }
+            logger.info("重新连接, 第" + (times++) + "次");
+            MqttPushClient mqttPushClient = mqttConfig.getMqttPushClient();
+            if (mqttPushClient != null && mqttPushClient.getMqttClient() != null && mqttPushClient.getMqttClient().isConnected()) {
+                logger.info("重连成功");
+                break;
+            }
+            logger.info("重连失败");
+            // 每隔10秒重试一次
+            try {
+                TimeUnit.SECONDS.sleep(10);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
         }
-        // int times = 1;
-        // while (client == null || !client.isConnected()) {
-        //     logger.info("重新连接, 第" + (times++) + "次");
-        //     MqttPushClient mqttPushClient = mqttConfig.getMqttPushClient();
-        //     if (mqttPushClient != null && mqttPushClient.getMqttClient() != null && mqttPushClient.getMqttClient().isConnected()) {
-        //         logger.info("重连成功");
-        //         break;
-        //     }
-        //     logger.info("重连失败");
-        //     // 每隔10秒重试一次
-        //     try {
-        //         TimeUnit.SECONDS.sleep(10);
-        //     } catch (InterruptedException e) {
-        //         e.printStackTrace();
-        //     }
-        // }
     }
 
     @Override
-    public void messageArrived(String topic, MqttMessage message) throws Exception {
-        // 收到消息并设置返回字符串格式
-        String payload = new String(message.getPayload(), "UTF-8");
+    public void messageArrived(String topic, MqttMessage message) {
+        try {
+            // 收到消息并设置返回字符串格式
+            String payload = new String(message.getPayload(), "UTF-8");
 
-        //logger.info("接收消息内容：payload格式:{}", payload);
+            //logger.info("接收消息内容：payload格式:{}", payload);
 
-        // 解析数据
-        DeviceInfo device = JSON.parseObject(payload, DeviceInfo.class);
+            // 解析数据
+            DeviceInfo device = JSON.parseObject(payload, DeviceInfo.class);
 
-        syncProperties(device);
+            syncProperties(device);
 
-        checkAlarm(device);
+            checkAlarm(device);
+        } catch (Exception e) {
+            logger.error("消息接收处理异常", e);
+        }
     }
 
     @Override