场景介绍
实现测试客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能。
MQTT发布消息到代理服务器
安装paho-mqtt
使用pip工具安装paho-mqtt，输入以下指令即可：
sudo pip install paho-mqtt
安装 MQTT 客户端库
为了方便连接到 MQTT 服务器，我们需要安装 paho-mqtt 库。可以选择以下两种方法之一进行安装。
1、使用源码安装
git clone https://github.com/eclipse/paho.mqtt.python
cd paho.mqtt.python
python3 setup.py install
2、使用 pip3 安装
pip3 install paho-mqtt
MQTT 的使用
连接 MQTT 服务器
可以使用 EMQX 提供的 免费公共 MQTT 服务器，该服务基于 EMQX 的 MQTT 物联网云平台 创建。服务器接入信息如下：
● Broker: broker.emqx.io
● TCP Port: 1883
● Websocket Port: 8083
如果有需要，也可以使用 docker 在本地快速安装 EMQX 服务器。
docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 18083:18083 emqx/emqx
连接测试代码
est_connect.py
import paho.mqtt.client as mqtt 回调函数。当尝试与 MQTT broker 建立连接时，触发该函数。
client 是本次连接的客户端实例。
userdata 是用户的信息，一般为空。但如果有需要，也可以通过 user_data_set 函数设置。
flags 保存服务器响应标志的字典。
rc 是响应码。一般情况下，我们只需要关注 rc 响应码是否为 0 就可以了。
def on_connect(client, userdata, flags, rc):if rc == 0:print("Connected success")else:print(f"Connected fail with code {rc}")client = mqtt.Client()
client.on_connect = on_connect
client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()
将上面的代码保存为 test_connect.py 文件，并运行
python3 test_connect.py
我们在 on_connect 函数里对响应码进行了判断，为 0 则输出 Connected success 表示连接成功。如果返回的是其它数字，我们就需要对照下面的响应码进行判断。
在 MQTT 协议的概念中，消息是通过主题传递的，比如设备 A 向主题 T 发送消息，那么只有订阅了主题 T 的设备才能接收到。所以仅仅接入 MQTT 服务器并没有太大意议，要完整地使用 MQTT 服务，我们还需要知道如何订阅和发布消息。
订阅消息
打开任意编辑器，输入下面的代码，并保存为 subscriber.py 文件：
#subscriber.py
import paho.mqtt.client as mqttdef on_connect(client, userdata, flags, rc):print(f"Connected with result code {rc}")# 订阅，需要放在 on_connect 里# 如果与 broker 失去连接后重连，仍然会继续订阅 raspberry/topic 主题client.subscribe("raspberry/topic")# 回调函数，当收到消息时，触发该函数
def on_message(client, userdata, msg):print(f"{msg.topic} {msg.payload}")client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message# 设置遗嘱消息，当树莓派断电，或者网络出现异常中断时，发送遗嘱消息给其他客户端
client.will_set('raspberry/status', b'{"status": "Off"}')# 创建连接，三个参数分别为 broker 地址，broker 端口号，保活时间
client.connect("broker.emqx.io", 1883, 60)# 设置网络循环堵塞，在调用 disconnect() 或程序崩溃前，不会主动结束程序
client.loop_forever()
调用 subscribe() 函数，可以让树莓派订阅一个主题。在上面的代码中，我们使用它订阅了 raspberry/topic 主题，并监听消息。
另外，我们还使用 will_set() 设置了遗嘱消息。 遗嘱消息是 MQTT 的一个特性，当设备在意外断开网络连接后，会向某个特定的主题发送消息。通过这个特性，可以得知树莓派是否断电，或者出现网络异常。
发布消息
打开任意编辑器，输入下面的代码，并保存为 publisher.py 文件：
import paho.mqtt.client as mqtt
import timedef on_connect(client, userdata, flags, rc):print(f"Connected with result code {rc}")client = mqtt.Client()
client.on_connect = on_connect
client.connect("broker.emqx.io", 1883, 60)
# 每间隔 1 秒钟向 raspberry/topic 发送一个消息，连续发送 5 次
for i in range(5):# 四个参数分别为：主题，发送内容，QoS, 是否保留消息client.publish('raspberry/topic', payload=i, qos=0, retain=False)print(f"send {i} to raspberry/topic")time.sleep(1)client.loop_forever()
调用 publish() 函数，可以向一个主题发送消息。在上面的代码中，我们使用了它向主题 raspberry/topic 发送消息。其中参数 QoS 是另一个 MQTT 特性，这里我们暂且设为 0。
MQTT客户端下载安装
浏览器输入网址：https://mqttx.app/zh下载MQTTX客户端工具进行测试。
测试订阅消息
运行 Python 代码，并主动发送一个消息。
-
打开终端，运行 Python 代码，监听消息 。
python3 subscriber.py -
使用 MQTTX 客户端与 MQTT 服务器建立连接，并向主题 raspberry/topic 发送消息 。
-
查看树莓派终端信息，将会看到已成功接收到 MQTTX 发布的消息。
测试发布消息
-
在 MQTTX 客户端中订阅 raspberry/topic 主题 。
-
在终端运行 Python 代码。
-
在 MQTTX 客户端中，查看树莓派发送的消息。
总结
我们完成了在树莓派上使用 Python MQTT 客户端库 paho-mqtt 编写测试客户端， 并实现了测试客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能。
