顾问头条

Python 客户端类库之paho-mqtt学习总结

2024-09-22 21:15 由 授客 发表于 #后端开发

实践环境

Python 3.9.13

paho-mqtt 2.1.0

简介

Eclipse Paho MQTT Python客户端类库实现了MQTT 协议版本 5.0, 3.1.1, 和3.1。

该类库提供一个客户端类，允许应用连接到MQTT代理并发布消息，订阅主题并检索发布的消息。同时还提供了一个写其它辅助函数，使向MQTT服务器发布一次性消息变得非常简单。

支持 Python 3.7+。

MQTT协议是一种机器对机器(M2M)/“物联网”连接协议。它被设计为一种极其轻量级的发布/订阅消息传输，对于需要小代码占用和/或网络带宽非常昂贵的远程连接非常有用。

安装

pip install paho-mqtt

已知限制

以下是已知的未实现的MQTT功能。

当clean_session为False时，会话仅存储在内存中，不会持久化。这意味着当客户端重新启动时（不仅仅是重新连接，通常是因为程序重新启动而重新创建对象），会话就会丢失。这可能会导致消息丢失。

客户端会话的以下部分丢失：

  • 已从服务器接收到但尚未完全确认的 QoS 2 消息。

    由于客户端会盲目确认任何PUBCOMP（QoS 2 事务的最后一条消息），因此它不会挂起，但会丢失此 QoS 2 消息。

  • 已发送到服务器但尚未完全确认的 QoS 1 和 QoS 2 消息。

    这意味着传递给 publish()的消息可能会丢失。这可以通过让传递给 publish() 的所有消息都有相应的on_publish() 调用或使用wait_for_publish来缓解。

    这也意味着代理在会话中可能有 QoS2 消息。由于客户端从一个空会话开始，它不知道它，并将重用mid。这还没有解决。

此外，当clean_session为True时，此类库将在网络重新连接时重新发布 QoS > 0消息。这意味着 QoS > 0消息不会丢失。但标准规定，我们应该丢弃发送发布包的任何消息。设置为True意味着不符合标准，QoS 2 可能会被接收两次。

如果只需要一次交付的 QoS 2 保证，则应设置clean_session=False。

用法与API

API详细在线文档：https://eclipse.dev/paho/files/paho.mqtt.python/html/client.html

示例：https://github.com/eclipse/paho.mqtt.python/tree/master/examples

开始

下面是一个非常简单的示例，它订阅代理$SYS主题树并打印出结果消息：

# -*- coding:utf-8 -*-

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, reason_code, properties):
    '''客户端从服务器接收到 CONNACK 响应时的回调'''

    print(f"Connected with result code {reason_code}")  # 成功连接时 reason_code 值为 Success

    # 在on_connect()中执行订阅操作，意味着如果应用失去连接并且重新连接后，订阅将被续订。
    if reason_code == 'Success':
        client.subscribe('$SYS/#')

def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f'Disconnected with result code {reason_code}')


def on_message(client, userdata, msg):
    '''从服务器收到 PUBLISH 消息时的回调。'''
    print(msg.topic + ' ' + str(msg.payload)) # 输出值形如 $SYS/broker/version b'mosquitto version 2.0.18'

mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.on_connect = on_connect
mqttc.on_disconnect = on_disconnect
mqttc.on_message = on_message
 
# client.username_pw_set('testacc', 'test1234') # 设置访问账号和密码

mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)

# 阻塞调用，处理网络流量、分派回调和处理重新连接
# 有其它提供线程接口和手动接口的loop*()函数可用
mqttc.loop_forever()

说明：

  1. Client.username_pw_set(username: str | None, password: str | None = None*) → None

    为代理身份验证设置用户名和密码(可选)。

    必须在connect()之前调用才能生效。需要支持MQTT v3.1或更高版本的代理。

    • 参数：
      • username – 要进行身份验证的用户名。需要与客户端id没有关系。必须是字符串[MQTT-3.1.3-11]。设置为“None”可将客户端重置为不使用用户名/密码进行代理身份验证。
      • password – 用于身份验证的密码。可选，如果不需要，则设置为None。如果为字符串r，那么它将被编码为UTF-8。
  2. Client.connect(host: str, port: int = 1883, keepalive: int = 60, bind_address: str = '', bind_port: int = 0, clean_start: bool | Literal[3] = 3, properties: Properties|None = None) → MQTTErrorCode

    连接到远程代理。这是一个阻塞调用，用于建立底层连接并传输CONNECT数据包。请注意，在收到并处理CONNACK之前，连接状态不会更新（这需要一个正在运行的网络循环，请参阅loop_start, loop_forever, loop…)).
    参数：

    • host– 远程代理的主机名或IP地址。

    • port– 要连接的服务器主机的网络端口。默认为1883。请注意，SSL/TLS上MQTT的默认端口是8883，因此如果使用TLS_set()可能需要提供端口。

    • keepalive - 设置心跳的时间，单位是秒。这个值告诉MQTT客户端，在没有接收到任何通信的情况下，多久应该发送一个PING请求给服务器，以保持连接，默认60秒。

    • clean_start -（仅限MQTT v5.0）True、False或MQTT_CLEAN_START_FIRST_ONLY。总是设置MQTT v5.0 clean_start标志、从不或仅在第一次成功连接时。设置clean_start标志后，MQTT会话数据（如未完成的消息和订阅）在成功连接时被清除。对于MQTT v3.1.1，Client的clean_session参数应用于类似的结果。

    • properties (Properties) –( 仅仅限MQTT v5.0)需要在MQTT连接包发送的的MQTT v5.0 属性。

客户端(Client)

Client类一般使用流程如下：

  1. 创建客户端实例
  2. 使用connect*() 函数之一连接到代理
  3. 调用其中一个loop*()函数来维护代理的网络流量
  4. 使用subscribe()订阅主题并接收消息
  5. 使用publish()将消息发布到代理
  6. 使用disconnect()断开与代理的连接

将调用回调以允许应用程序根据需要处理事件。这些回调如下所述。

网络循环

这些功能是Client背后的驱动力。如果它们没有被调用，传入的网络数据将不会被处理，传出的网络数据也不会被发送。管理网络环路有四种选择。这里描述了三个，第四个在下面的“外部事件循环支持”中描述。不要混合使用不同的loop函数。

loop_start() / loop_stop()
mqttc.loop_start()

while True:
    temperature = sensor.blocking_read()
    mqttc.publish("paho/temperature", temperature)

mqttc.loop_stop()

这些函数实现了网络循环的线程接口。在connect*()之前或之后调用loop_start()一次，会在后台运行一个线程来自动调用loop()。这释放了主线程，用于可能阻塞的其他工作。此调用还处理与代理的重新连接。调用loop_stop() 以停止后台线程。如果调用disconnect()，循环也会停止。

loop_forever()
mqttc.loop_forever(retry_first_connection=False)

这是网络循环的阻塞形式，在客户端调用disconnect()之前不会返回(即调用mqttc.disconnect()后会停止阻塞，继续运行其后的代码)。它会自动处理重新连接。
除了使用connect_async时的第一次连接尝试外，使用retry_first_connection=True 使其重试第一次连接。
警告：这可能会导致客户端保持连接到不存在的主机而不会出现失败。

loop()
run = True
while run:
    rc = mqttc.loop(timeout=1.0)
    if rc != 0:
        # need to handle error, possible reconnecting or stopping the application

定期调用以处理网络事件。此调用触发select()等待，直到网络套接字可用于读取或写入，如果套接字可用，则处理流入/流出的数据。此函数最多阻塞timeout秒。timeout不能超过客户端的keepalive值，否则代理会定期断开客户端的连接。
使用这种循环，需要自己处理重新连接策略。

回调

与paho-mqtt交互的接口包括各种回调，当发生某些事件时，类库会调用这些回调。

回调是在代码中定义的函数，用于实现对这些事件要求的操作。这可能只是打印收到的消息，也可能是更复杂的行为。

回调API是有版本的，所选版本是我们提供给客户端构造函数的CallbackAPIVersion。目前支持两个版本：

  • CallbackAPIVersion.VERSION1：这是paho-mqtt 2.0版本之前使用的历史版本。它是在引入CallbackAPIVersion之前使用的API。此版本已弃用，将在paho-mqtt 3.0版本中删除。
  • CallbackAPIVersion.VERSION2：此版本在协议MQTT 3.x和MQTT 5.x之间更为一致。它也更适用于MQTT 5.x，因为reason_code和属性始终在可获取时提供。建议所有用户升级到此版本。强烈建议MQTT 5.x用户使用。

存在以下回调：

  • on_connec()：当收到代理返回CONNACK时被调用。调用可能是针对被拒绝的连接，请检查reason_code以查看连接是成功还是被拒绝。
  • on_connect_fail()：当TCP连接建立失败时，由loop_forever()和loop_start()调用。当直接使用connect()或reconnect()时，不会调用此回调。它仅由loop_start()和loop_forever()制造的自动（重新）连接后被调用
  • on_disconnect()：当连接关闭时被调用。
  • on_message()：收到代理返回的MQTT消息时被调用。
  • on_publish()：当MQTT消息发送到代理时被调用。取决于QoS级别，回调在不同时刻被调用：
    • 对于QoS==0，一旦消息通过网络发送，就会调用它。这可能是在相应的publish()返回之前。
    • 对于QoS==1，当收到代理返回的对应消息的PUBACK时调用它
    • 对于QoS==2，当收到代理返回的对应消息的PUBCOMP时，会调用它
  • on_subscribe()：当收到代理返回的SUBACK时被调用
  • on_unsubscribe：当收到代理返回的UNSUBACK时被调用
  • on_log()：当类库记录一条消息时被调用
  • onSocket_open、onSocket_close、onSocket_register_write、onSocket_unregister_write：用于外部循环支持(External event loop support)的回调。详见下文。

参阅在线文档查看有关每个回调的特征。

订阅示例
# -*- coding:utf-8 -*-

import paho.mqtt.client as mqtt

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    # 由于我们只订阅了一个信道，reason_code_list只包含一个条目
    # print(reason_code_list) #输出： [ReasonCode(Suback, 'Granted QoS 0')]
    if reason_code_list[0].is_failure:
        print(f"Broker rejected you subscription: {reason_code_list[0]}")
    else:
        print(f"Broker granted the following QoS: {reason_code_list[0].value}")

def on_unsubscribe(client, userdata, mid, reason_code_list, properties):
    #注意，reason_code_list仅存在于MQTTv5中，在MQTTv3中，它将始终为空
    if len(reason_code_list) == 0 or not reason_code_list[0].is_failure:
        print("unsubscribe succeeded (if SUBACK is received in MQTTv3 it success)")
    else:
        print(f"Broker replied with failure: {reason_code_list[0]}")
    client.disconnect()

def on_message(client, userdata, message):
    # userdata是我们选择提供的数据结构，这里为一个列表(通过下方的 mqttc.user_data_set([])设置，该函数参数即为userdata参数值
    userdata.append(message.payload)
    # 假设只想处理10条消息
    if len(userdata) >= 10:
        client.unsubscribe("$SYS/#")

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Failed to connect: {reason_code}. loop_forever() will retry connection")
    else:
        # 应该始终在 on_connect 回调中订阅以确保在重新连接时订阅依旧存在。
        client.subscribe("$SYS/#")

mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.on_subscribe = on_subscribe
mqttc.on_unsubscribe = on_unsubscribe

mqttc.user_data_set([]) # 设置 userdata
mqttc.connect("mqtt.eclipseprojects.io")
mqttc.loop_forever() # 当调用client.disconnect()后继续执行以下代码

print(f"Received the following message: {mqttc.user_data_get()}")
发布示例
# -*- coding:utf-8 -*-

import time
import paho.mqtt.client as mqtt

def on_publish(client, userdata, mid, reason_code, properties):
    '''reason_code和properties将仅出现在MQTTv5中。在MQTTv3中始终未设置
    使用不存在`uncaked_publish`中的`mid`调用`on_publish()`。这是由于不可避免的竞争情形：
    * publish() 返回已发送消息的mid。
    * 主线程将publish()返回的mid添加到uncaked_publish中
    * loop_start线程调用on_publish()
    虽然不太可能（因为on_publish()将在网络往返后调用），但是这是一种可能发生的竞争情形
    避免竞争情形的最佳解决方案是使用publish()中的msg_info。还可以尝试使用已确认的mid列表，而不是从待处理列表中删除
    但是请记住，mid可以重复使用！
    reason_code和properties将仅出现在MQTTv5中。在MQTTv3中始终未设置
    '''
    try:
        userdata.remove(mid)
    except KeyError:
        print("on_publish() is called with a mid not present in unacked_publish")
        print("This is due to an unavoidable race-condition:")
        print("* publish() return the mid of the message sent.")
        print("* mid from publish() is added to unacked_publish by the main thread")
        print("* on_publish() is called by the loop_start thread")
        print("While unlikely (because on_publish() will be called after a network round-trip),")
        print(" this is a race-condition that COULD happen")
        print("")
        print("The best solution to avoid race-condition is using the msg_info from publish()")
        print("We could also try using a list of acknowledged mid rather than removing from pending list,")
        print("but remember that mid could be re-used !")

unacked_publish = set()
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.on_publish = on_publish

mqttc.user_data_set(unacked_publish)
mqttc.connect("mqtt.eclipseprojects.io")
mqttc.loop_start()

# 应用生产一些消息
msg_info = mqttc.publish("paho/test/topic", "my message", qos=1)
unacked_publish.add(msg_info.mid)

msg_info2 = mqttc.publish("paho/test/topic", "my message2", qos=1)
unacked_publish.add(msg_info2.mid)

# 等待所有消息被发布
while len(unacked_publish):
    time.sleep(0.1)

# 由于上述描述的竞争状态, 以下等待所有消息发布完成的方式更安全
msg_info.wait_for_publish()
msg_info2.wait_for_publish()

mqttc.disconnect()
mqttc.loop_stop()

说明：

  1. Client.max_inflight_messages_set(inflight: int) → None

    设置一次可以通过其网络流的QoS>0的消息的最大数量(可以简单理解为允许多大数量的QoS>0的消息被同时进行传输处理)。默认值为20。

  2. Client.max_queued_messages_set（queue_size:int）→ Client
    设置传出消息队列中的最大消息数量。0表示无限制。

  3. MQTTMessageInfo.wait_for_publish(timeout: float | None = None) → None

    阻塞，直到与此对象关联的消息被发布，或者直到超时发生。如果timeout为None，则永远不会超时。将超时设置为正数秒，例如1，2，以启用超时。
    抛出：

    • ValueError–如果消息因传出队列已满而未排队。
    • RuntimeError-如果消息因其他原因未发布。

  4. 实践过程中发现，采用多线程并发发布消息时，如果服务器因为限流的原因不返回消息确认，那么运行一小段时间后，出现消息无法发布成功的情况(不报错，但是消息无法抵达broker)，通过合理的参数调用以上三个函数，可以缓解这个问题。

Logger

客户端会发出一些日志消息，这些消息在故障排除过程中可能很有用。启用日志最简单的方法是调用enable_logger()。可以提供自定义记录器或使用默认记录器

示例:

import logging
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.DEBUG)

mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.enable_logger()

mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)
mqttc.loop_start()

# Do additional action needed, publish, subscribe, ...
[...]

还可以定义一个on_log回调，它将接收所有日志消息的副本。例子：

import paho.mqtt.client as mqtt

def on_log(client, userdata, paho_log_level, messages):
    if paho_log_level == mqtt.LogLevel.MQTT_LOG_ERR:
        print(message)

mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.on_log = on_log

mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)
mqttc.loop_start()

# Do additional action needed, publish, subscribe, ...
[...]

Paho日志级别和标准日志级别的对应关系如下：

Paho logging
MQTT_LOG_ERR logging.ERROR
MQTT_LOG_WARNING logging.WARNING
MQTT_LOG_NOTICE logging.INFO (no direct equivalent)
MQTT_LOG_INFO logging.INFO
MQTT_LOG_DEBUG logging.DEBUG

外部事件循环支持

为了支持其他网络循环，如asyncio（参见示例)，类库公开了一些方法和回调来支持这些用例。

存在以下循环方法：

  • loop_read：应该在套接字可读取时调用。
  • loop_write：应该在套接字可写并且类库需要写入数据时调用。
  • loop_misc：应每隔几秒钟调用一次，以处理消息重试和ping。

用伪代码表示如下:

while run:
    if need_read:
        mqttc.loop_read()
    if need_write:
        mqttc.loop_write()
    mqttc.loop_misc()

    if not need_read and not need_write:
        # But don't wait more than few seconds, loop_misc() need to be called regularly
        wait_for_change_in_need_read_or_write()
    updated_need_read_and_write()

棘手的部分是实现updated_need_read_and_write并等待条件变更。为了支持这一点，存在以下方法：

  • socket：当TCP连接打开时返回socket对象。此调用对于基于select循环特别有用。请参阅examples/loop_select.py 。

  • want_write()：如果有数据等待写入，则返回True。这接近于上述伪代码的need_writew，但还是应该检查套接字是否可写。

  • 回调函数on_socket_*：

    • on_socket_open：在套接字打开时调用。
    • on_socket_open：在套接字打开时调用。
    • on_socket_close：当套接字即将关闭时调用。
    • on_socket_register_write：当客户端想要在套接字上写入数据时调用
    • on_socket_unregister_write：当套接字上没有更多数据要写入时调用。

    回调对于事件循环特别有用，在事件循环中，可以注册或注销用于读写的套接字。请参阅examples/loop_asyncio.py 获取示例。

回调总是按以下顺序调用：

  • on_socket_open

  • 0或者更多次：

    • on_socket_register_write
    • on_socket_unregister_write

  • on_socket_close

全局辅助函数

客户端模块还提供了一些全局辅助函数。

topic_matches_sub(sub, topic)可用于检查主题(topic)是否与订阅(subscription)匹配。
例如：

主题foo/bar 将与订阅foo/#或+/bar匹配
主题non/matching 将不匹配订阅non/+/+

发布

此模块提供了两个辅助函数single()和multiple()，允许以一次性方式直接发布消息。换句话说，它们对于有一个/多个消息要发布到代理，然后断开连接而不需要其他任何东西的情况非常有用。

提供的两个函数是single()和multiple()。

这两个函数都支持MQTT v5.0，但目前不允许在连接或发送消息时设置任何属性。

Single

发布一条消息到代理，然后彻底断开连接。

例子：

import paho.mqtt.publish as publish

publish.single("paho/test/topic", "payload", hostname="mqtt.eclipseprojects.io")

Multiple

发布多条消息到代理，然后彻底断开连接。

例子：

from paho.mqtt.enums import MQTTProtocolVersion
import paho.mqtt.publish as publish

msgs = [{'topic':"paho/test/topic", 'payload':"multiple 1"},
    ("paho/test/topic", "multiple 2", 0, False)]
publish.multiple(msgs, hostname="mqtt.eclipseprojects.io", protocol=MQTTProtocolVersion.MQTTv5)

订阅

此模块提供了两个辅助函数simple()和callback()，以允许直接订阅和处理消息。

这两个函数都支持MQTT v5.0，但目前不允许在连接或发送消息时设置任何属性。

Simple

订阅一组主题并返回收到的消息。这是一个阻塞函数。
例子：

import paho.mqtt.subscribe as subscribe

msg = subscribe.simple("paho/test/topic", hostname="mqtt.eclipseprojects.io")
print("%s %s" % (msg.topic, msg.payload))

使用回调(Callback)

订阅一组主题，并使用用户提供的回调处理收到的消息。

例子：

import paho.mqtt.subscribe as subscribe

def on_message_print(client, userdata, message):
    print("%s %s" % (message.topic, message.payload))
    userdata["message_count"] += 1
    if userdata["message_count"] >= 5:
        # it's possible to stop the program by disconnecting
        client.disconnect()

subscribe.callback(on_message_print, "paho/test/topic", hostname="mqtt.eclipseprojects.io", userdata={"message_count": 0})

参考连接

https://github.com/eclipse/paho.mqtt.python

https://eclipse.dev/paho/files/paho.mqtt.python/html/client.html