Python 是一种广泛使用的解释型、高级编程、通用型编程语言。Python 的设计哲学强调代码的可读性和简洁的语法(尤其是使用空格缩进划分代码块&＃xff0c;而非使用大括号或者关键词)。Python 让开发者能够用更少的代码表达想法&＃xff0c;不管是小型还是大型程序&＃xff0c;该语言都试图让程序的结构清晰明了。

MQTT 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 &＃xff0c;可以用极少的代码和带宽为联网设备提供实时可靠的消息服务&＃xff0c;它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。

本文主要介绍如何在 Python 项目中使用 paho-mqtt 客户端库 &＃xff0c;实现客户端与 MQTT 服务器的连接、订阅、取消订阅、收发消息等功能。

项目初始化

本项目使用 Python 3.6 进行开发测试&＃xff0c;读者可用如下命令确认 Python 的版本。

➜ ~ python3 --version Python 3.6.7

选择 MQTT 客户端库

paho-mqtt 是目前 Python 中使用较多的 MQTT 客户端库&＃xff0c;它在 Python 2.7 或 3.x 上为客户端类提供了对 MQTT v3.1 和 v3.1.1 的支持。它还提供了一些帮助程序功能&＃xff0c;使将消息发布到 MQTT 服务器变得非常简单。

Pip 安装 Paho MQTT 客户端

Pip 是 Python 包管理工具&＃xff0c;该工具提供了对 Python 包的查找、下载、安装、卸载的功能。

pip3 install -i https://pypi.doubanio.com/simple paho-mqtt

Python MQTT 使用

连接 MQTT 服务器

本文将使用 EMQ X 提供的 免费公共 MQTT 服务器 &＃xff0c;该服务基于 EMQ X 的 MQTT 物联网云平台 创建。服务器接入信息如下&＃xff1a;

• Broker: broker.emqx.io
• TCP Port: 1883
• Websocket Port: 8083

导入 Paho MQTT客户端

from paho.mqtt import client as mqtt_client

设置 MQTT Broker 连接参数

设置 MQTT Broker 连接地址&＃xff0c;端口以及 topic&＃xff0c;同时我们调用 Python random.randint 函数随机生成 MQTT 客户端 id。

broker &＃61; &＃39;broker.emqx.io&＃39;port &＃61; 1883topic &＃61; "/python/mqtt"client_id &＃61; f&＃39;python-mqtt-{random.randint(0, 1000)}&＃39;

编写 MQTT 连接函数

编写连接回调函数 on_connect &＃xff0c;该函数将在客户端连接后被调用&＃xff0c;在该函数中可以依据 rc 来判断客户端是否连接成功。通常同时我们将创建一个 MQTT 客户端&＃xff0c;该客户端将连接到 broker.emqx.io 。

def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc &＃61;&＃61; 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d", rc) # Set Connecting Client ID client &＃61; mqtt_client.Client(client_id) client.on_connect &＃61; on_connect client.connect(broker, port) return client

发布消息

首先定义一个 while 循环语句&＃xff0c;在循环中我们将设置每秒调用 MQTT 客户端 publish 函数向 /python/mqtt 主题发送消息。

def publish(client): msg_count &＃61; 0 while True: time.sleep(1) msg &＃61; f"messages: {msg_count}" result &＃61; client.publish(topic, msg) # result: [0, 1] status &＃61; result[0] if status &＃61;&＃61; 0: print(f"Send &＃96;{msg}&＃96; to topic &＃96;{topic}&＃96;") else: print(f"Failed to send message to topic {topic}") msg_count &＃43;&＃61; 1

订阅消息

编写消息回调函数 on_message &＃xff0c;该函数将在客户端从 MQTT Broker 收到消息后被调用&＃xff0c;在该函数中我们将打印出订阅的 topic 名称以及接收到的消息内容。

def subscribe(client: mqtt_client): def on_message(client, userdata, msg): print(f"Received &＃96;{msg.payload.decode()}&＃96; from &＃96;{msg.topic}&＃96; topic") client.subscribe(topic) client.on_message &＃61; on_message

完整代码

消息发布代码

# python 3.6import randomimport timefrom paho.mqtt import client as mqtt_clientbroker &＃61; &＃39;broker.emqx.io&＃39;port &＃61; 1883topic &＃61; "/python/mqtt"# generate client ID with pub prefix randomlyclient_id &＃61; f&＃39;python-mqtt-{random.randint(0, 1000)}&＃39;def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc &＃61;&＃61; 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d", rc) client &＃61; mqtt_client.Client(client_id) client.on_connect &＃61; on_connect client.connect(broker, port) return clientdef publish(client): msg_count &＃61; 0 while True: time.sleep(1) msg &＃61; f"messages: {msg_count}" result &＃61; client.publish(topic, msg) # result: [0, 1] status &＃61; result[0] if status &＃61;&＃61; 0: print(f"Send &＃96;{msg}&＃96; to topic &＃96;{topic}&＃96;") else: print(f"Failed to send message to topic {topic}") msg_count &＃43;&＃61; 1def run(): client &＃61; connect_mqtt() client.loop_start() publish(client)if __name__ &＃61;&＃61; &＃39;__main__&＃39;: run()

消息订阅代码

# python3.6import randomfrom paho.mqtt import client as mqtt_clientbroker &＃61; &＃39;broker.emqx.io&＃39;port &＃61; 1883topic &＃61; "/python/mqtt"# generate client ID with pub prefix randomlyclient_id &＃61; f&＃39;python-mqtt-{random.randint(0, 100)}&＃39;def connect_mqtt() -> mqtt_client: def on_connect(client, userdata, flags, rc): if rc &＃61;&＃61; 0: print("Connected to MQTT Broker!") else: print("Failed to connect, return code %d", rc) client &＃61; mqtt_client.Client(client_id) client.on_connect &＃61; on_connect client.connect(broker, port) return clientdef subscribe(client: mqtt_client): def on_message(client, userdata, msg): print(f"Received &＃96;{msg.payload.decode()}&＃96; from &＃96;{msg.topic}&＃96; topic") client.subscribe(topic) client.on_message &＃61; on_messagedef run(): client &＃61; connect_mqtt() subscribe(client) client.loop_forever()if __name__ &＃61;&＃61; &＃39;__main__&＃39;: run()

测试

消息发布

运行 MQTT 消息发布代码&＃xff0c;我们将看到客户端连接成功&＃xff0c;并且成功将消息发布。

python3 pub.py

消息订阅

运行 MQTT 消息订阅代码&＃xff0c;我们将看到客户端连接成功&＃xff0c;并且成功接收到发布的消息。

python3 sub.py

总结

至此&＃xff0c;我们完成了使用 paho-mqtt 客户端连接到 公共 MQTT 服务器 &＃xff0c;并实现了测试客户端与 MQTT 服务器的连接、消息发布和订阅。

与 C &＃43;&＃43; 或 Java 之类的高级语言不同&＃xff0c;Python 比较适合设备侧的业务逻辑实现&＃xff0c;使用 Python 您可以减少代码上的逻辑复杂度&＃xff0c;降低与设备的交互成本。我们相信在物联网领域 Python 将会有更广泛的应用。

原文作者:EMQ

原文链接:https://www.emqx.io/cn/blog/how-to-use-mqtt-in-python