前言

使用python编写程序进行测试MQTT的发布和订阅功能。首先要安装:pip install paho-mqtt

测试发布（pub）

我的MQTT部署在阿里云的服务器上面，所以我在本机上编写了python程序进行测试。

然后在shell里面重新打开一个终端，订阅一个主题为“chat” mosquitto_sub -t chat

在本机上测试远程的MQTT的发布功能就是把自己作为一个发送信息的人，当自己发送信息的时候，所有订阅过该主题（topic）的对象都将收到自己发送的信息。
mqtt_client.py

# encoding: utf-8import paho.mqtt.client as mqttHOST = "101.200.46.138"PORT = 1883def test():    client = mqtt.Client()    client.connect(HOST, PORT, 60)    client.publish("chat","hello liefyuan",2) # 发布一个主题为'chat',内容为‘hello liefyuan’的信息    client.loop_forever()if __name__ == '__main__':    test()1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

注解函数：

client.connect(self, host, port, keepalive, bind_address)

client.publish(self, topic, payload, qos, retain)

client.subscribe(self, topic, qos)

测试订阅（sub）

在本机上编写程序测试订阅功能，就是让自己的程序作为一个接收者，同一个主题没有发布（pub）信息的时候，就自己一直等候。

# encoding: utf-8import paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc):    print("Connected with result code "+str(rc))    client.subscribe("chat")def on_message(client, userdata, msg):    print(msg.topic+" " + ":" + str(msg.payload))client = mqtt.Client()client.on_connect = on_connectclient.on_message = on_messageclient.connect("www.liefyuan.top", 1883, 60)client.loop_forever()1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。