实现mqtt断线重连

iot-common/iot-common-driver/src/main/java/com/shkpr/iot/common/driver/config/DriverConfig.java

@@ -0,0 +1,16 @@
+package com.shkpr.iot.common.driver.config;
+
+import com.shkpr.iot.common.driver.properties.DriverProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * 驱动配置
+ *
+ * @author 欧阳劲驰
+ * @since 0.0.1-dev
+ */
+@Configuration
+@EnableConfigurationProperties(DriverProperties.class)
+public class DriverConfig {
+}

iot-common/iot-common-driver/src/main/java/com/shkpr/iot/common/driver/properties/DriverProperties.java

@@ -0,0 +1,23 @@
+package com.shkpr.iot.common.driver.properties;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * 驱动属性
+ *
+ * @author 欧阳劲驰
+ * @since 0.0.1-dev
+ */
+@Getter
+@Setter
+@ConfigurationProperties(prefix = "driver")
+public class DriverProperties {
+    /**
+     * 检查间隔
+     */
+    private Duration checkInterval;
+}

iot-driver/iot-driver-mqtt/src/main/java/com/shkpr/iot/driver/mqtt/service/MqttService.java

@@ -3,19 +3,16 @@ package com.shkpr.iot.driver.mqtt.service;
 import com.shkpr.iot.common.core.domain.po.DataInfo;
 import com.shkpr.iot.common.core.domain.po.Destination;
 import com.shkpr.iot.common.core.domain.po.Equipment;
+import com.shkpr.iot.common.driver.properties.DriverProperties;
 import com.shkpr.iot.common.driver.service.DriverService;
 import com.shkpr.iot.driver.mqtt.common.context.MqttHandlerContext;
 import com.shkpr.iot.driver.mqtt.handler.MqttInboundHandler;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.eclipse.paho.client.mqttv3.IMqttActionListener;
-import org.eclipse.paho.client.mqttv3.IMqttToken;
-import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
-import org.eclipse.paho.client.mqttv3.MqttClient;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
-import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -23,6 +20,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -45,11 +43,22 @@ public class MqttService implements DriverService {
      * <p>{@code k:code,v:client}</p>
      */
     private final static Map<String, ScheduledFuture<?>> SCHEDULED_CACHE = new ConcurrentHashMap<>(2048);
+
+    final
+    ThreadPoolTaskScheduler taskScheduler;
+    final
+    DriverProperties properties;
+
     /**
      * 数据消费
      */
     private Consumer<DataInfo> dataConsumer;
 
+    public MqttService(ThreadPoolTaskScheduler taskScheduler, DriverProperties properties) {
+        this.taskScheduler = taskScheduler;
+        this.properties = properties;
+    }
+
     /**
      * 获取url
      *
@@ -71,6 +80,20 @@ public class MqttService implements DriverService {
         equipments.stream()
                 .filter(equipment -> equipment != null && equipment.getEnable())
                 .forEach(this::register);
+        //定时重新注册
+        final AtomicBoolean firstSkip = new AtomicBoolean(true);
+        taskScheduler.scheduleWithFixedDelay(() -> {
+            //跳过首次执行
+            if (firstSkip.getAndSet(false)) return;
+            //遍历设备
+            equipments.forEach(equipment -> {
+                //如不是全部连接,则重新注册
+                if (!equipment.getDestinations().stream().allMatch(this::isConnected)) {
+                    refRegister(equipment);
+                    log.info("监测到MQTT目标断开,重新注册,设备:{}", equipment);
+                }
+            });
+        }, properties.getCheckInterval());
     }
 
     /**
@@ -119,6 +142,8 @@ public class MqttService implements DriverService {
 
     /**
      * 注册
+     *
+     * @param equipment 设备
      */
     @Override
     public void register(Equipment equipment) {
@@ -137,11 +162,24 @@ public class MqttService implements DriverService {
     }
 
     /**
-     * {@inheritDoc}
+     * 重新注册
+     *
+     * @param equipment 设备
      */
     @Override
     public void refRegister(Equipment equipment) {
+        if (equipment.getEquipmentCode() == null) return;
+        //获取设备标识
+        String code = equipment.getEquipmentCode();
+        if (StringUtils.isBlank(code) || !equipment.getEnable()) return;
+        //断开连接,并重连
+        equipment.getDestinations().forEach(destination -> {
+            disconnect(destination);
+            connect(equipment, destination);
+        });
 
+        //重新启动任务
+        if (!equipment.getPassive()) refSchedule(equipment);
     }
 
     /**
@@ -174,6 +212,7 @@ public class MqttService implements DriverService {
                 public void onSuccess(IMqttToken asyncActionToken) {
                     try {
                         mqttClient.subscribe(destination.getTopic(), 2);
+                        log.info("MQTT连接成功,目标:{}", destination);
                     } catch (MqttException e) {
                         log.error("MQTT客户端关闭失败,通讯标识:{},{}", destination.getHost(), destination.getPort(), e);
                     }
@@ -202,7 +241,10 @@ public class MqttService implements DriverService {
     }
 
     /**
-     * {@inheritDoc}
+     * 是否连接
+     *
+     * @param destination 目标
+     * @return 连接状态
      */
     @Override
     public Boolean isConnected(Destination destination) {

iot-server/iot-server-data/src/main/resources/application.yml

@@ -4,5 +4,10 @@ spring:
   profiles:
     active:
       - baokang
-
+  task:
+    scheduling:
+      pool:
+        size: 4
+driver:
+  check-interval: PT1M