Python 客户端类库之paho-mqtt学习总结
实践环境
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()
说明:
-
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。
- 参数:
-
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.0clean_start
标志、从不或仅在第一次成功连接时。设置clean_start标志后,MQTT会话数据(如未完成的消息和订阅)在成功连接时被清除。对于MQTT v3.1.1,Client
的clean_session
参数应用于类似的结果。 -
properties
(Properties) –( 仅仅限MQTT v5.0)需要在MQTT连接包发送的的MQTT v5.0 属性。
-
客户端(Client)
Client类一般使用流程如下:
- 创建客户端实例
- 使用
connect*()
函数之一连接到代理 - 调用其中一个
loop*()
函数来维护代理的网络流量 - 使用
subscribe()
订阅主题并接收消息 - 使用
publish()
将消息发布到代理 - 使用
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()
说明:
-
Client.max_inflight_messages_set(inflight: int) → None
设置一次可以通过其网络流的QoS>0的消息的最大数量(可以简单理解为允许多大数量的QoS>0的消息被同时进行传输处理)。默认值为20。
-
Client.max_queued_messages_set(queue_size:int)→ Client
设置传出消息队列中的最大消息数量。0表示无限制。 -
MQTTMessageInfo.wait_for_publish(timeout: float | None = None) → None
阻塞,直到与此对象关联的消息被发布,或者直到超时发生。如果
timeout
为None
,则永远不会超时。将超时设置为正数秒,例如1,2,以启用超时。
抛出:ValueError
–如果消息因传出队列已满而未排队。RuntimeError
-如果消息因其他原因未发布。
-
实践过程中发现,采用多线程并发发布消息时,如果服务器因为限流的原因不返回消息确认,那么运行一小段时间后,出现消息无法发布成功的情况(不报错,但是消息无法抵达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