如何在 Python 中使用 MQTT

2023-01-09 19:24:14 浏览数 (2)

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

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

项目初始化

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

代码语言:javascript复制
➜  ~ python3 --version             
Python 3.6.7
选择 MQTT 客户端库

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

Pip 安装 Paho MQTT 客户端

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

代码语言:javascript复制
pip3 install -i https://pypi.doubanio.com/simple paho-mqtt

Python MQTT 使用

连接 MQTT 服务器

本文将使用 EMQ X 提供的 免费公共 MQTT 服务器 (https://www.emqx.io/cn/mqtt/public-mqtt5-broker),该服务基于 EMQ X 的 MQTT 物联网云平台 (https://cloud.emqx.io/) 创建。服务器接入信息如下:

  • Broker: broker.emqx.io
  • TCP Port: 1883
  • Websocket Port: 8083
导入 Paho MQTT客户端
代码语言:javascript复制
from paho.mqtt import client as mqtt_client
设置 MQTT Broker 连接参数

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

代码语言:javascript复制
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

代码语言:javascript复制
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 %dn", 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 主题发送消息。

代码语言:javascript复制
 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 收到消息后被调用,在该函数中我们将打印出订阅的 topic 名称以及接收到的消息内容。

代码语言:javascript复制
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

完整代码

消息发布代码

代码语言:javascript复制
# 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 %dn", 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()

消息订阅代码

代码语言:javascript复制
# 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 %dn", 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 消息发布代码,我们将看到客户端连接成功,并且成功将消息发布。

代码语言:javascript复制
python3 pub.py
消息订阅

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

代码语言:javascript复制
python3 sub.py

总结

至此,我们完成了使用 paho-mqtt 客户端连接到 公共 MQTT 服务器 (https://www.emqx.io/cn/mqtt/public-mqtt5-broker) ,并实现了测试客户端与 MQTT 服务器的连接、消息发布和订阅。

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

0 人点赞