mqtt接收数据12

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/subscribe/PushCallback.java

@@ -71,7 +71,7 @@ public class PushCallback implements MqttCallback {
             System.out.println(jsonObject1);
             try {
                 PublishMQTT publishMQTT = new PublishMQTT();
-                publishMQTT.start("tcp://115.238.57.190:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
+                publishMQTT.start("tcp://115.238.57.190:1883", "publish", "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
             } catch (Exception e) {
                 System.out.println(e);
             }
@@ -108,7 +108,7 @@ public class PushCallback implements MqttCallback {
             jsonObject1.put("alarm_cfg", map);
             System.out.println(jsonObject1);
             PublishMQTT publishMQTT = new PublishMQTT();
-            publishMQTT.start("tcp://115.238.57.190:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
+            publishMQTT.start("tcp://115.238.57.190:1883", "publish", "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
             log.info("----报警配置下发完成---");
         } else if ("get_dev_ctrl".equals(messages)) {
             JSONObject jsonObject = new JSONObject();
@@ -131,7 +131,7 @@ public class PushCallback implements MqttCallback {
             jsonObject.put("dev_ctrl", i);
             log.info("设备配置完成开始发送：" + i);
             PublishMQTT publishMQTT = new PublishMQTT();
-            publishMQTT.start("tcp://115.238.57.190:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject.toJSONString());
+            publishMQTT.start("tcp://115.238.57.190:1883", "publish", "admin", "admin", "huimv_down_" + chip, jsonObject.toJSONString());
             log.info("---设备控制发送完成---");
         }  else if ("dev_ctrl_suc".equals(messages)) {
             System.out.println("收到控制响应");

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/uplod/UploadMqtt.java

@@ -1,6 +1,7 @@
 package com.huimv.env.manage.saas.mqtt.uplod;
 
 
+
 import com.huimv.env.manage.saas.mqtt.subscribe.PushCallback;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -9,7 +10,9 @@ import lombok.SneakyThrows;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import org.springframework.transaction.annotation.Transactional;
+
+
+import java.util.List;
 
 @EqualsAndHashCode(callSuper = true)
 @Data
@@ -21,14 +24,17 @@ public class UploadMqtt extends Thread {
 
     private String userName = "admin";    //非必须
     private String passWord = "admin";  //非必须
-    private String topic ;
+    private List<String> topics ;
+
 
     @SneakyThrows
     @Override
     public synchronized void  run() {
         try {
+
+
             // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
-            client = new MqttClient(HOST, clientid, new MemoryPersistence());
+            client = new MqttClient(HOST, "subscribe", new MemoryPersistence());
             // MQTT的连接设置
             options = new MqttConnectOptions();
             // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
@@ -50,7 +56,11 @@ public class UploadMqtt extends Thread {
 
             //订阅消息
 //            int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
-            client.subscribe(topic,1);
+            for (String topic : topics) {
+                System.out.println("走到了这里开始订阅！");
+                client.subscribe("huimv_up_"+topic, 1);
+            }
+//            client.subscribe(topic,1);
         } catch (Exception e) {
             e.printStackTrace();
         }

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/uplod/UploadMqttStart.java

@@ -6,6 +6,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
+import java.util.ArrayList;
 import java.util.List;
 
 @Component
@@ -17,13 +18,16 @@ public class UploadMqttStart {
     @PostConstruct
     public  void  test() {
         List<EnvDeviceRegister> list = deviceRegisterService.list();
+        System.out.println("设备的数量："+list.size());
+
+        List<String> chipIds = new ArrayList<>();
         for (EnvDeviceRegister deviceRegister : list) {
             String chipId = deviceRegister.getChipId();
-            UploadMqtt uploadmqtt = new UploadMqtt();
-            uploadmqtt.setClientid("pub_" + chipId);
-            uploadmqtt.setTopic("huimv_up_" + chipId);
-            uploadmqtt.start();
+            chipIds.add(chipId);
         }
+        UploadMqtt uploadmqtt = new UploadMqtt();
+        uploadmqtt.setTopics(chipIds);
+        uploadmqtt.start();
     }
 
 }