α-LIB
0%

python实现MQTT的订阅和发布

发表于 更新于 分类于

安装依赖库

1
pip install paho-mqtt

封装订阅的部分，形成公用函数 (subscribe.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import threading
import paho.mqtt.client as mqtt
import time

HOST = "xxxxx" #emq服务器地址
PORT = 1883

class Mqtt_subscribe(threading.Thread):
    """
    mqtt thread, 完成订阅功能
    """

    def __init__(self, subtopic):
        super(Mqtt_subscribe, self).__init__()
        self.client_id = time.strftime(
            '%Y%m%d%H%M%S', time.localtime(
                time.time()))
        self.client = mqtt.Client(self.client_id)
        self.client.user_data_set(subtopic)
        self.client.username_pw_set("admin", "public")
        self.message = None

    def run(self):
        # ClientId不能重复，所以使用当前时间
        # 必须设置，否则会返回「Connected with result code 4」
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        self.client.connect(HOST, PORT, 60)
        self.client.loop_forever(timeout=60)

    def on_connect(self, client, subtopic, flags, rc):
        print("Connected with result code " + str(rc))
        print("topic:" + subtopic)
        client.subscribe(subtopic, 2)

    def on_message(self, client, userdata, msg):
        # print(msg.topic + " " + msg.payload.decode("utf-8"))
        self.mess_age = msg.payload.decode("utf-8")
        if 'test' in self.mess_age:   #根据订阅内容判断，如果含有想要的值则会停止订阅
            self.client.disconnect()
        return self.mess_age

if __name__ == "__main__":
    subtopic = "test"
    t = Mqtt_subscribe(subtopic)
    t.start()

封装发布的部分 (mqtt_publish.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import paho.mqtt.publish as publish
import time

def mqtt_Publish(topic,msg):
    client_id = time.strftime('%Y%m%d%H%M%S', time.localtime(time.time()))
    publish.single(topic,msg,qos=0,
                   hostname="xxxx", #emq地址
                   port=1883, 
                   client_id=client_id,
                   auth={'username': "admin", 'password': "public"}) #登录密码，官方是amdin,public

if __name__ == '__main__':
    topic = 'test'
    message = "this is a test"
    mqtt_Publish(topic,message)

以下面使用为例，使用订阅和发布（received.py,先运行该文件，然后往该主题中发布订阅 - 工具或者 mqtt_publish.py 文件都可以，即可把订阅到的变量值取出用于断言）

1
2
3
4
5
6
7
8
9
10
11
from testcase.test_04 import subscribe

topic = 'test'
mess = subscribe.Mqtt_subscribe(topic)
mess.start()
threads =[]
threads.append(mess)
for t in threads:
    t.join(20) #超时20S不会再线程阻塞，如果没有接受到内容需要一直阻塞，则t.join()即可
a = mess.mess_age
print("这是变量信息",a)
相关文章