在 Python 中使用 MQTT 的方法
本文详细介绍如何在 Python 项目中利用 paho-mqtt 库连接 MQTT 服务器。通过配置 broker.emqx.io 公共服务器，演示了客户端 ID 生成、连接回调、消息发布及订阅的完整流程，并提供基于 Python 3.6 的发布与订阅代码示例及测试结果。
Python 是一门广泛应用的解释型高级通用编程语言。其设计理念注重代码可读性与语法简洁性，特别是通过空格缩进来界定代码块，而非依赖大括号或特定关键词。这种设计使得开发者能够以精简的代码表达复杂逻辑，无论是小型脚本还是大型应用，都能保持结构清晰。
MQTT 是一种基于发布/订阅模式的轻量级物联网消息传输协议。该协议以极低的代码量和带宽消耗，为联网设备提供实时且可靠的消息服务，目前广泛应用于物联网、移动互联网、智能硬件、车联网以及电力能源等多个行业领域。
本文重点讲解如何在 Python 项目中集成 paho-mqtt 客户端库，从而实现客户端与 MQTT 服务器之间的连接管理、主题订阅与取消、以及消息的收发操作。
项目初始化
本项目的开发与测试环境基于 Python 3.6，读者可通过执行特定命令来确认当前 Python 的版本信息。
➜ ~ python3 --version
Python 3.6.7选择 MQTT 客户端库
paho-mqtt 是当前 Python 生态中较为流行的 MQTT 客户端库，它支持 Python 2.7 及 3.x 版本，并为客户端类提供了对 MQTT v3.1 和 v3.1.1 协议的支持。此外，该库还提供了一系列辅助功能，极大地简化了向 MQTT 服务器发布消息的过程。
Pip 安装 Paho MQTT 客户端
Pip 是 Python 官方推荐的包管理工具，具备查找、下载、安装及卸载 Python 包的完整功能，是管理第三方库的标准方式。
pip3 install -i https://pypi.doubanio.com/simple paho-mqttPython MQTT 使用
连接 MQTT 服务器
本文示例将采用 EMQ X 提供的免费公共 MQTT 服务器，该服务依托于 EMQ X 的 MQTT 物联网云平台构建。具体的服务器接入参数如下：
- Broker: broker.emqx.io
- TCP Port: 1883
- Websocket Port: 8083
导入 Paho MQTT客户端
from paho.mqtt import client as mqtt_client设置 MQTT Broker 连接参数
需配置 MQTT Broker 的连接地址、端口号以及订阅主题，同时利用 Python 的 random.randint 函数随机生成一个唯一的 MQTT 客户端 ID。
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
client_id = f'python-mqtt-{random.randint(0, 1000)}'编写 MQTT 连接函数
需要编写连接回调函数 on_connect，该函数在客户端成功连接后被触发。在此函数中，可通过返回值 rc 判断连接是否成功。通常，我们会创建一个 MQTT 客户端实例，并将其配置为连接到 broker.emqx.io。
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
# Set Connecting Client ID
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client发布消息
首先定义一个 while 无限循环，在循环体内设置每秒调用 MQTT 客户端的 publish 函数，向 /python/mqtt 主题持续发送消息。
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1订阅消息
编写消息回调函数 on_message，当客户端从 MQTT Broker 接收到消息时触发。在该函数中，我们将打印出所订阅的主题名称以及接收到的具体消息内容。
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
client.subscribe(topic)
client.on_message = on_message完整代码
消息发布代码
# python 3.6
import random
import time
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
if __name__ == '__main__':
run()消息订阅代码
# python 3.6
import random
import time
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 1000)}'
def connect_mqtt():
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def publish(client):
msg_count = 0
while True:
time.sleep(1)
msg = f"messages: {msg_count}"
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"Send `{msg}` to topic `{topic}`")
else:
print(f"Failed to send message to topic {topic}")
msg_count += 1
def run():
client = connect_mqtt()
client.loop_start()
publish(client)
if __name__ == '__main__':
run()
消息订阅代码
# python3.6
import random
from paho.mqtt import client as mqtt_client
broker = 'broker.emqx.io'
port = 1883
topic = "/python/mqtt"
# generate client ID with pub prefix randomly
client_id = f'python-mqtt-{random.randint(0, 100)}'
def connect_mqtt() -> mqtt_client:
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
else:
print("Failed to connect, return code %d\n", rc)
client = mqtt_client.Client(client_id)
client.on_connect = on_connect
client.connect(broker, port)
return client
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
client.subscribe(topic)
client.on_message = on_message
def run():
client = connect_mqtt()
subscribe(client)
client.loop_forever()
if __name__ == '__main__':
run()测试
消息发布
运行 MQTT 消息发布代码后，预期将看到客户端连接成功的提示，并且消息能够成功发布至服务器。
python3 pub.py
消息订阅
运行 MQTT 消息订阅代码后，预期将看到客户端连接成功的提示，并且能够成功接收到之前发布的消息。
python3 sub.py
总结
至此，我们完成了使用 paho-mqtt 客户端连接公共 MQTT 服务器的全过程，并成功测试了客户端与服务器之间的连接建立、消息发布及订阅功能。
与 C++ 或 Java 等高级语言相比，Python 更适合用于设备侧的业务逻辑实现。使用 Python 可以有效降低代码的逻辑复杂度，从而减少与硬件设备的交互成本。预计在未来物联网领域，Python 将获得更广泛的应用。
后续我们将陆续发布更多关于物联网开发及 Python 技术的相关文章，敬请持续关注。
以上就是在 Python 中使用 MQTT 的方法的详细内容，更多关于 Python 中使用 MQTT 的资料请关注本站其它相关文章！
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
