mqtt

src/main/java/com/care/bms/mqtt/MqttDataConnector.java

@@ -45,9 +45,9 @@ public class MqttDataConnector {
         mqttConnectOptions.setCleanSession(false);
         // mqttConnectOptions.sto
         // 设置超时时间 s
-        mqttConnectOptions.setConnectionTimeout(20);
+        mqttConnectOptions.setConnectionTimeout(30);
         // 设置会话心跳时间
-        mqttConnectOptions.setKeepAliveInterval(10);
+        mqttConnectOptions.setKeepAliveInterval(60);
         mqttConnectOptions.setAutomaticReconnect(true);//设置自动重连
         //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
         // mqttConnectOptions.setWill("sec", "close".getBytes(), 2, true);
@@ -87,9 +87,6 @@ public class MqttDataConnector {
 
 
     /**
-     * 完成连接时 主要用于断开连接时 重新订阅
-     * 网上说 将 mqttConnectOptions.setCleanSession(false); 就可以继续接收，试了下没有效果，用的是开源版
-     * 的emq测试的，不知道企业版是否有效
      *
      * testTopic/#   #多层通配符    +单层通配符
      * topicFilter
@@ -98,25 +95,17 @@ public class MqttDataConnector {
      * @throws Exception
      */
     public void setReSubscribe(String topicFilter, MqttMessageListener iotMqttMessageListener) throws Exception {
-        //先订阅一次
-        mqttClient.subscribe(topicFilter,iotMqttMessageListener);
-        //断线重连
+        int qos = 2;//0：最多一次 、1：最少一次 、2：只有一次
+        mqttClient.subscribe(topicFilter,qos, iotMqttMessageListener);
         mqttClient.setCallback(new MqttCallbackExtended() {
             @Override
             public void connectComplete(boolean reconnect, String serverURI) {
-                if(reconnect){
-                    try {
-                        mqttClient.subscribe(topicFilter,iotMqttMessageListener);
-                        logger.info("mqtt重新建立连接后,topic={} 重新订阅！",topicFilter);
-                    } catch (MqttException e) {
-                        e.printStackTrace();
-                    }
-                }
+                logger.info("reconnect ---> " + reconnect + "       serverURI--->" + serverURI);
             }
 
             @Override
             public void connectionLost(Throwable cause) {
-
+                logger.error("connectionLost cause ---> " + cause);
             }
 
             @Override

src/main/java/com/care/bms/mqtt/MqttMessageListener.java

@@ -94,10 +94,6 @@ public class MqttMessageListener implements IMqttMessageListener {
 
         String mqttMessageStr = mqttMessage.toString();
         System.out.println("mqttMessageStr == " + mqttMessageStr);
-        String saveDatabase = CommonConfUtil.getConf("mqtt.save.database");
-        if(!"1".equals(saveDatabase)){
-            return;
-        }
         if(mqttMessageStr == null || !mqttMessageStr.startsWith("{")){
             return;
         }