发送emqq

emq_terminal.json

@@ -3,7 +3,7 @@
         
         "emq_host": "139.224.69.89",
         "emq_port": 31101,
-        "serial_no": "VT00001",
+        "serial_no": "WF00000001",
         "measurements": [
             {
                 "measurement": "door",
@@ -18,7 +18,7 @@
 
         "emq_host": "139.224.69.89",
         "emq_port": 31101,
-        "serial_no": "VT00002",
+        "serial_no": "WF00000002",
         "measurements": [
             {
                 "measurement": "door",
@@ -33,7 +33,7 @@
 
         "emq_host": "139.224.69.89",
         "emq_port": 31101,
-        "serial_no": "VT00003",
+        "serial_no": "WF00000003",
         "measurements": [
             {
                 "measurement": "door",

fake_terminal.py

 # encoding=utf8
 import json
 import random
-import sys
-from paho.mqtt import client as mqtt_client
-
+import threading
 ########################
 # 虚拟终端上报数据脚本
 ########################
 import time
 
-client_id = f'python-mqtt-{random.randint(0, 1000)}'
-topic = "wftest"
+from paho.mqtt import client as mqtt_client
+
+client_id = f''
+topic = "/a1hjFmQV0zo"
 username = "admin"
 password = "public"
 
@@ -22,7 +22,7 @@ cmdTemplate = """
         "telemetry": {telemetry}
     }},
     "time": {time},
-    "mqttTopic": "/{topic}/{devSn1}/lalala",
+    "mqttTopic": "{topic}/{devSn1}/user/group/data",
     "type": "data",
     "deviceName": "105104012101126339",
     "group": "saaspri",
@@ -62,9 +62,7 @@ def getTelemetry(param):
 
 def connect_mqtt(ip, port):
     def on_connect(client, userdata, flags, rc):
-        if rc == 0:
-            print("Connected to MQTT Broker!")
-        else:
+        if rc != 0:
             print("Failed to connect, return code %d\n", rc)
 
     client = mqtt_client.Client(client_id)
@@ -72,37 +70,46 @@ def connect_mqtt(ip, port):
 
     client.on_connect = on_connect
     client.connect(ip, port)
+
     return client
 
 
-def publish(client, msg):
-    result = client.publish(topic, msg)
+def publish(client, msg, topicSn):
+    result = client.publish(topicSn, msg)
     # result: [0, 1]
     status = result[0]
     if status == 0:
-        print(f"Send `{msg}` to topic `{topic}`")
+        print(f"Send `{msg}` to topic `{topicSn}`")
     else:
-        print(f"Failed to send message to topic {topic}")
+        print(f"Failed to send message to topic {topicSn}")
 
 
-def run(ip, port, msg):
+def run(ip, port, msg, topicSn):
     client = connect_mqtt(ip=ip, port=port)
     client.loop_start()
-    publish(client, msg)
+    publish(client, msg, topicSn)
+
+
+def test(data):
+    while True:
+        tel = getTelemetry(data.get("measurements"))
+        cmd = cmdTemplate.format(devSN=data.get("serial_no"), telemetry=tel, time=str(time.time()),
+                                 devSn1=data.get("serial_no"), topic=topic)
+        cmdJson = json.loads(cmd)
+        topicSn = cmdJson.get("mqttTopic")
+        # print(cmd)
+        run(data.get("emq_host"), data.get("emq_port"), cmd, topicSn)
+        time.sleep(10)
 
 
 if __name__ == '__main__':
 
     jsonFile = "emq_terminal.json"
-    while(True):
-        with open(jsonFile, "r") as fp:
-            datas = json.load(fp)
-            dataLen = len(datas)
-            for key in range(dataLen):
-                data = datas[key]
-                tel = getTelemetry(data.get("measurements"))
-                cmd = cmdTemplate.format(devSN=data.get("serial_no"), telemetry=tel, time=str(time.time()),
-                                         devSn1=data.get("serial_no"), topic=topic)
-                print(cmd)
-                run(data.get("emq_host"), data.get("emq_port"), cmd)
-        time.sleep(10)
+    with open(jsonFile, "r") as fp:
+        datas = json.load(fp)
+        dataLen = len(datas)
+        for key in range(dataLen):
+            t = threading.Thread(target=test, args=(datas[key],))
+            t.start()
+    while True:
+        time.sleep(100)