Version 1.0.4:

1. 修改MqttListener支持断线重连后重新订阅checkIn主题.

build.gradle

@@ -15,7 +15,7 @@ apply plugin: 'idea'
 apply plugin: 'org.springframework.boot'
 
 group = 'com.example'
-version = '1.0.3-release'
+version = '1.0.4-release'
 sourceCompatibility = 1.8
 
 repositories {

src/main/java/com/example/tdl/mqtt/MqttListener.java

@@ -37,7 +37,7 @@ import java.util.concurrent.TimeUnit;
  **/
 
 @Component
-public class MqttListener implements MqttCallback {
+public class MqttListener implements MqttCallbackExtended {
 
     private Logger logger = LoggerFactory.getLogger(MqttListener.class);
     private static final int INIT_DELAY_DEFAULT = 1000; // unit:ms
@@ -69,14 +69,15 @@ public class MqttListener implements MqttCallback {
     @PostConstruct
     public void initialize() throws MqttException {
         try {
+            String clientId = "TDL_CheckIn" + "_" + String.valueOf(System.currentTimeMillis()) + "_sub";
             mqttAsyncClient = new MqttTemlateAsync(mqttconfig.getUrl() + ":" + mqttconfig.getPort(),
-                    "TDL_CheckIn" + "_" + String.valueOf(System.currentTimeMillis()) + "_sub");
+                    clientId);
             mqttAsyncClient.connect(MqttTemlateAsync.setSSLOptions(mqttconfig.getUsername(), mqttconfig.getPassword(),
                     mqttconfig.getCacrt(), mqttconfig.getClientcrt(),mqttconfig.getClientkey(), mqttconfig.getClientpwd()), null,
                     new IMqttActionListener() {
                         @Override
                         public void onSuccess(IMqttToken iMqttToken) {
-                            logger.info("connection successfull ");
+                            logger.info("connect successfully!");
                             try {
                                 mqttAsyncClient.subscribe(listener_topic, mqttconfig.getQos());
                             } catch (MqttException e) {
@@ -86,7 +87,13 @@ public class MqttListener implements MqttCallback {
 
                         @Override
                         public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
-                            logger.error("connection fail with client id");
+                            logger.error("connection fail(" + clientId + "), reason is " + throwable.getMessage());
+                            try {
+                                logger.info("Start to reconnect(" + clientId + ")");
+                                mqttAsyncClient.reconnect();
+                            } catch (MqttException e) {
+                                logger.error("MqttAsyncService reconnect MqttException: " + e.getMessage());
+                            }
                         }
                     });
             this.mqttAsyncClient.setCallback(this);
@@ -95,38 +102,24 @@ public class MqttListener implements MqttCallback {
         }
     }
 
-    private void clientReconnect() {
-        if (mReconnectScheduler != null) {
-            // mReconnectScheduler is running.
-            return;
-        }
-
-        mReconnectScheduler = Executors.newSingleThreadScheduledExecutor();
-        mReconnectScheduler.scheduleAtFixedRate(new Runnable() {
-            @Override
-            public void run() {
-                // TODO Auto-generated method stub
-                if (mqttAsyncClient != null && !mqttAsyncClient.isConnected()) {
+    private void resubscribeTopics() {
         try {
-                        mqttAsyncClient.reconnect();
-                    } catch (MqttSecurityException e) {
-                        // TODO: handle exception
-                        e.printStackTrace();
+            mqttAsyncClient.subscribe(listener_topic, mqttconfig.getQos());
         } catch (MqttException e) {
-                        // TODO: handle exception
-                        e.printStackTrace();
-                    }
-                } else {
-                    mReconnectScheduler.shutdown();
-                    mReconnectScheduler = null;
+            logger.error("connection MqttException: " + e.getMessage());
         }
     }
-        }, mInitDelay, mSchedulePeriod, TimeUnit.MILLISECONDS);
+
+    @Override
+    public void connectComplete(boolean reconnect, String serverURI) {
+        logger.info((reconnect ? "reconnect" : "connect") + " complete!");
+        /* connect/reconnected complete, then subscribe topics stored */
+        resubscribeTopics();
     }
 
     @Override
     public void connectionLost(Throwable cause) {
-        System.out.println(cause);
+        logger.info("connection lost, reason is " + cause.getMessage());
         //   clientReconnect();
     }
 
@@ -288,8 +281,6 @@ public class MqttListener implements MqttCallback {
         }
     }
 
-
-
     //判断两个list是否一致
     public synchronized <T extends Comparable<T>> boolean compare(List<T> a, List<T> b) {
         if (a.size() != b.size()){
@@ -301,6 +292,4 @@ public class MqttListener implements MqttCallback {
         }
         return true;
     }
-
-
 }
\ No newline at end of file