Python中MQTT

Python有许多优秀的MQTT客户端,比较有代表性的有paho-mqtt、hbmqtt、gmqtt等，各有特色

• paho-mqtt 有着最优秀的文档，代码风格易于理解，同时有着强大的基金会支持,目前新版本支持 MQTT 5.0

• hbmqtt 使用 asyncio 库实现，可以优化网络 I/O 带来的延迟,但是代码风格不友好，文档较少,不支持 MQTT 5.0,且不再维护,被原作者弃用,有一个分支amqtt正在由不同的人积极开发

• gmqtt 同样通过 asyncio 库实现，相比 HBMQTT ，代码风格友好，最重要的是支持 MQTT 5.0

  paho-mqtt 可以说是 Python MQTT 开源客户端库中的佼佼者， 支持5.0、3.1.1 和 3.1 的MQTT 协议,由 Eclipse 基金会主导开发.在基金会的支持下，以每年一个版本的速度更新,稳定、持续,本文使用新版本paho-mqtt v1.6.1

  pypi地址

  开源地址

  参考文章

  • http://www.steves-internet-guide.com/ (墙外)

paho-mqtt安装

使用pip安装

pip3 install paho-mqtt
# virtualenv安装和完整代码参考官方文档

paho-mqtt已知的一些限制

截止1.6.1版本,当 clean_session 为 False 时，session 只存储在内存中，不持久化。这意味着当客户端重新启动时(不仅仅是重新连接，通常是因为程序重新启动而重新创建对象)会话丢失。这可能导致消息丢失。

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

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

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

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

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

  这也意味着代理可能在会话中有 Qos2 消息。由于客户端从一个空会话开始，它不知道它并将重新使用中间。这还没有确定。

此外，当 clean_session 为 True 时，该库将在网络重新连接时重新发布 QoS > 0 消息。这意味着 QoS > 0 消息不会丢失。但是标准规定我们是否应该丢弃发送了发布数据包的任何消息。我们的选择意味着我们不符合标准，并且 QoS 2 有可能被接收两次。如果只需要一次交付的 QoS 2 保证，应该 clean_session = False

paho-mqtt使用

使用paho-mqtt实现客户端相关功能简单步骤如下:

• 构造Client客户端实例
• 使用connect相关方法将创建的客户端连接到代理
• 使用loop相关方法维护和broker的通信
• 使用subscribe()方法订阅主题、接收消息
• 使用publish()方法发送消息
• 使用disconnect()断开连接

---

Client客户端

使用客户端连接代理、订阅等，首先我们需要先创建一个客户端,paho-mqtt使用Client()创建客户端实例

Client类的构造参数

# Client 源码 参数如下
def __init__(self, client_id="", clean_session=None, userdata=None,
           protocol=MQTTv311, transport="tcp", reconnect_on_failure=True):

Client类构造参数讲解

# 参数示意
client_id = '客户端ID,str类型,可自定义'
'''
如果长度为零或者为空,则会随机生成一个,这种情况下，clean_session参数必须为True清除会话,因为持久会话需要client_id为唯一标识来恢复会话
'''

clean_session = '会话清除状态,bool类型,设为True,broker在断开连接的时候删除该client的信息,为False,则相当于是持久会话'
'''
clean_session在官方文档示例默认值为True,查看1.6.1版本源码默认值为None,看逻辑应该是支持mqttv5之后做的的变化
源码中判断如果protocol是MQTT V5版本 clean_session不是None的话则抛出ValueError,就是mqtt5的clean_session必须持久会话,
如果不是mqtt5的话如果clean_session is None 则将clean_session设置为True,这块就与官方文档的默认逻辑对应上了
'''

userdata = 'client用户数据,传递给回调函数，可以是任意类型,可以使用Clinet的 user_data_set()函数进行更新数据'

protocol = '客户端协议的版本,默认是MQTTv311就是3.1.1版本,也可以是MQTTv31、MQTTv5版本'
'''
protocol的参数在源码中是以下对应关系，理论上直接传入对应int值或者导入MQTTv** 字段传入都可
MQTTv31 = 3
MQTTv311 = 4
MQTTv5 = 5
'''

transport = '底层传输协议,默认是使用原始tcp,值可以设置为websockets通过ws发送mqtt'

reconnect_on_failure = 'bool类型, 连接失败后是否自动重新connect，默认为True'
'''
官方文档没有reconnect_on_failure相关示例和解释,不确定老版本有没有该字段
'''

创建客户端

# -*- coding: utf-8 -*-
# @Time:     2023/5/10 16:09
# @Author:   LiQi
# @Describe:

import paho.mqtt.client as mqtt  # 导入clinet 别名 mqtt

# 创建一个客户端实例赋值client,client_id自定义,其他参数根据需要设定
client = mqtt.Client(client_id='muziqi')

重置客户端

'''paho-mqtt提供reinitialise方法重新初始化已经构造好的客户端'''
# 上面创建的客户端clinet,直接调用reinitialise
client.reinitialise()
'''
reinitialise方法拥有下面三个参数
client_id="", clean_session=True, userdata=None
'''

Clinet选项函数

选项函数的作用是给构造好的客户端添加附加选项,一般情况下在使用content连接broker前完成,比如设置Client的证书、回调、账号密码等,根据具体业务使用,设置完成后使用conten连接到broker

max_inflight_messages_set

当QoS> 0的时候，可以存在的部分完成(已经发送，还没有确认成功的消息)的消息的最大数量,这些消息可以同时通过网络流,默认为20.增加此值将消耗更多内存，但可以增加吞吐量

# inflight参数为要修改的数量,最小为1,小于1则抛出ValueError
client.max_inflight_messages_set(inflight)

max_queued_messages_set

设置传出消息队列中等待处理的QoS> 0的传出消息的最大数量,默认为0表示无限制，当队列已满时，其他传出的消息都将被丢弃，实际使用中0实际上并不是无限,目前实现最大可为65555,包括队列中的65535条消息+inflight的默认20条消息，如果inflight默认值被消息,队列中的实际消息数量被减少

# queue_size是要修改的最大数量
client.max_queued_messages_set(queue_size)

message_retry_set-废弃

QoS>0 的消息,如果发送之后超过一定时间broker没有响应,重发消息的时间,单位是s，默认5s

'''
 源码注释该方法不再使用，在2.0版本删除
 '''

ws_set_options

设置WebSocket的连接选项,构造Client时transport="websockets"才会使用,connect相关方法之前调用

import paho.mqtt.client as mqtt

# 传输协议设置为ws
client = mqtt.Client(client_id='muziqi',transport='websockets')

'''
path:broker的mqtt 路径,以/开头的字符串
headers:头部信息可以是字典或者是可调用对象。如果是字典的话字典中额外的项被添加到websocket头中。如果是一个可调用的对象，默认的websocket的头部信息被传递到这个函数，将函数结果当做新的头不新鲜'''
client.ws_set_options(path="/mqtt", headers=None)

tls_set

配置网络加密和身份验证选项,启用SSL/TLS支持,connect相关方法之前调用

# 以下所有参数默认值均为None,根据业务自身选用
client.tls_set(ca_certs, certfile, keyfile, cert_reqs, tls_version, ciphers, keyfile_password)
'''
ca_certs:  CA 根证书的路径,如果这是给定的唯一选项，则客户端将以与 Web 浏览器类似的方式运行,要求代理拥有由ca_certs中的证书颁发机构签名的证书，并将使用 TLS v1.2 进行通信，但不会尝试任何形式的身份验证。这提供了基本的网络加密，具体取决于代理的配置方式。默认情况下，在 Python 2.7.9+ 或 3.4+ 上，使用系统的默认证书颁发机构。在较旧的 Python 版本上，此参数是必需的

certfile和keyfile: 客户端私钥和证书的路径,分别指向 PEM 编码的客户端证书和私钥的字符串,如果这些参数不是None那么它们将被用作基于 TLS 的身份验证的客户端信息,对此功能的支持取决于代理

cert_reqs: 设置客户端对 broker 证书的需求，默认是 ssl.CERT_REQUIRED ，表示 broker 必须提供一个证书

tls_version:使用的SSL/TLS协议版本,默认是TLS v1.2 一般不做修改

ciphers:字符串,设置允许使用哪些加密方法,默认是None

keyfile_password:如果certfile和keyfile加密了需要密码解密,可以使用该参数传递,如果不提供的话需要在终端输入,目前无法定义回调以提供密码

'''

tls_set_context

配置网络加密和身份验证上下文,启用SSL/TLS支持,connect相关方法之前调用

client.tls_set_context(context)

'''
参数释义
ssl.SSLContext对象,在python3.4之后,默认情况下由ssl.create_default_context()提供,一般无需自定义设置，业务需要不确定是否使用该方法的话,用tls_set or 默认上下文即可
'''

tls_insecure_set

配置对服务器证书中的服务器主机名进行验证,在connect相关方法之前和 tls_set 或 tls_set_context之后调用

client.tls_insecure_set(value)
'''
value是bool类型,设置为True代币不使用加密,Flase使用加密
值设置为True，可能让恶意第三方通过 DNS 欺骗等方式冒充你的服务器
测试程序可以使用该方法来不进行验证
'''
# 源码根据tls_set选项的cert_reqs字段来设置默认值
if cert_reqs != ssl.CERT_NONE:

    self.tls_insecure_set(False)
else:

    self.tls_insecure_set(True)

enable_logger

使用 python 日志包 logging启用日志记录,可以跟后面的on_log日志回调方法同时使用

client.enable_logger(logger)
'''
logger参数用来指定记录器
如果指定了记录器，那么将使用该logging.Logger对象，否则将自动创建一个
'''

Paho日志记录级别和标准日志级别对应关系如下

disable_logger

禁用python 日志包 记录日志,对on_log回调没有影响

client.disable_logger()
'''
disable_logger相当于是把enable_logger记录器清除
两个方法的源码也很简单 
		↓↓↓
'''
def enable_logger(self, logger=None):
    if logger is None:
        if self._logger is not None:
            return
        logger = logging.getLogger(__name__)
    self._logger = logger

def disable_logger(self):
    self._logger = None

username_pw_set

设置用户名和可选的代理身份验证密码,密码根据broker验证进行设定,可为空,在connect相关方法前调用

client.username_pw_set(username,password=None)

user_data_set

将在生成事件时传递给回调方法的私有用户数据,可以使用这个方法更新构造Client时候的userdata字段值

client.user_data_set(userdata)

will_set

设置遗嘱消息,client没有使用disconnect方法以外断开连接,broker推送遗嘱消息

client.will_set(topic,payload,qos,retain,properties)
'''
topic:遗嘱发布主题
payload:默认值为None,遗嘱消息内容
qos:默认值为0,消息质量级别
retain:bool,默认False,是否将遗嘱消息设置为保留消息
properties:默认值为None，支持MQTTv5后新增的字段，用来设置遗嘱属性
'''

遗嘱消息逻辑、报文、上述字段、属性可参考https://www.cnblogs.com/Mickey-7/p/17385599.html>

reconnect_delay_set

客户端断开重新连接延迟的设置

client.reconnect_delay_set(min_delay,max_delay)
'''
min_delay:默认1
max_delay:默认值120
连接丢失的时候,默认等待min_delay，每次尝试重连后下次min_delay都将翻倍,上限时间是max_delay
连接成功将重置min_delay，Client收到broker的connack报文视为完全连接成功
'''

Client连接/重新连接/断开连接

---

连接方法-connect

connect方法的作用是将构造好、设置好选项的的client连接到broker,这是一个阻塞函数

client.connect(host, port, keepalive, bind_address, bind_port,clean_start,properties)

'''
host:远程代理的主机名或 IP 地址

prot:要连接的服务器主机的端口,默认为1883,如果是基于SSL/TLS的MQTT的默认端口为8883,如果选项函数使用tls_set或者tls_set_context，可能需要手动提供端口

keepalive:心跳间隔时间,默认为60，单位是秒

bind_address: 默认值是空字符串,如果client的本地有多个地址,可以使用该参数绑定一个地址,表示来源

bind_port:默认值是0， 与bind_address一样，本地有多个地址,可以使用该参数绑定一个端口,表示来源

clean_start:会话设置,默认值 MQTT_CLEAN_START_FIRST_ONLY = 3, mqtt v5版本仅在第一次成功连接时使用干净启动标志,mqttv5如果不是3则ValueError

properties:设置会话属性
'''

clean_start和properties参考：https://www.cnblogs.com/Mickey-7/p/17361919.html

连接方法-connect_async

异步连接方法,与loop_start结合使用以及非阻塞的方式连接,如果一直没有调用loop_start，不会完成连接

client.connect_async(host, port, keepalive, bind_address, bind_port,clean_start,properties)
'''
参数参考connect方法
'''

连接方法-connect_srv

DNS连接,使用 SRV DNS查找连接到代理以获取代理地址

client.connect_srv(domain, keepalive, bind_address,clean_start, properties)

'''
domain:用于搜索SRV记录的DNS域;如果没有则尝试本地域名
其余参数参考connect方法

'''

重新连接-reconnect

在使用connect相关方法连接过之后,想要重新连接可以使用reconnect方法,完全复用之前connect相关的参数进行连接

client.reconnect()

断开连接-disconnect

主动彻底断开和broker的连接,调用该方法不会等待排队的消息被发送

client.disconnect()

网络循环

网络循环的函数有四种,运行在后台，处理收发的消息,如果不使用的话，传入的数据不会被处理，传出的数据不会发送

loop

定期调用用以处理消息,通过 select() 函数阻塞，直到有消息需要收发，并根据消息类型触发对应的回调函数

run_status = True:
while run_status:
	clinet.loop(timeout,max_packets)
  
'''
timeout:默认值1.0,阻塞超时时间，单位S,不能超过心跳时间keepalive，否则client会定时从broker 断开
max_packets:默认值是1，不用设置，已经过时不使用
'''

loop_start / loop_stop

实现loop的调用和停止，在connect相关方法之前或之后调用loop_start，会在后台运行一个线程字段调用loop,无需编码实现loop循环,

每个客户端必须有一个loop循环

# 在connect相关方法之前或之后调用,会在后台运行一个线程调用loop,连接中断的时候会自动尝试重新连接,无需编码实现loop循环
client.loop_start()
# 停止loop循环的后台线程
client.loop_stop()

loop_forever

网络循环的阻塞形式,loop_forever 调用会阻塞主线程,永远不会停止,直到客户端调用disconnect,会自动重新连接

client.loop_forever(timeout,max_packets,retry_first_connection)
'''
retry_first_connection:bool,默认False,在第一次连接尝试失败后是否应该进行重试,与reconnect_on_failure不同，它只影响第一次连接,如果设置为True，当第一次连接失败时，paho-mqtt会抛出OSError/WebsocketConnectionError异常
'''

loop和loop_forever的区别

• loop方法：默认调用一下只循环一次，然后返回。这意味着它只会执行一次通信循环，不使用start方法的话需要无限while 并且在完成后将返回到调用程序。

• loop_forever方法：它是一个无限循环，如果没有意外情况，将一直保持运行。 它在连接丢失或出现其他错误时会自动重新连接。因此，它可以用于长期运行的应用程序，以保持MQTT客户端连接

loop_start和loop_forever的区别

• loop_start函数是一个非阻塞的函数，可以启动一个线程来处理MQTT客户端的网络通信和事件处理。该函数返回后，MQTT客户端将会在后台线程中运行，不会阻塞当前线程。在调用loop_start函数后，可以分别调用connect、publish、subscribe等方法来进行MQTT通信。但是在使用完MQTT客户端后，需要调用loop_stop函数来停止后台线程

• loop_forever函数是一个阻塞的函数，可以启动一个线程来处理MQTT客户端的网络通信和事件处理。该函数会一直阻塞当前线程，直到MQTT客户端接收到disconnect消息或者调用了disconnect函数。在调用loop_forever函数后，不能再使用connect、publish、subscribe等方法来进行MQTT通信，因为当前线程已经被阻塞。只有在调用了disconnect函数后，才会解除阻塞

• 因此，loop_start和loop_forever的区别在于是否阻塞当前线程。如果需要在MQTT客户端后台运行并且继续使用当前线程进行其他操作，可以使用loop_start函数；如果需要一直阻塞当前线程直到MQTT客户端退出，可以使用loop_forever函数

发布消息

publish

客户端发送消息到broker,broker将消息转发到订阅匹配主题的客户端

client.publish(topic, payload, qos, retain, properties)

'''
topic:消息发布的主题
payload:默认None,要发送的实际消息,如果没有或者为None，将发送长度为0的消息,payload传递int或者float会被默认转换为str发送真正类型的int、float需要使用struct.pack()创建 
qps:消息之类默认0
retain:是否设置保留消息
properties:设置保留消息的属性
'''

发布消息的返回

消息发送之后会返回一个MQTTMessageInfo对象,有以下属性和方法

• rc：发布的结果。可以是MQTT_ERR_SUCCESS表示成功，MQTT_ERR_NO_CONN表示客户端当前未连接，或者当使用 max_queued_messages_set时，MQTT_ERR_QUEUE_SIZE表示消息既没有排队也没有发送
• mid:发布请求的消息ID。 mid值可以通过检查on_publish()回调中的mid参数来跟踪发布请求，如果已定义，更容易使用wait_for_publish
• wait_for_publish() :将阻塞直到消息被发布。如果消息未排队(rc == MQTT_ERR_QUEUE_SIZE)，将引发 ValueError，如果发布时出现错误，则引发 RuntimeError，很可能是由于客户端未连接
• is_published:如果消息已发布返回True。如果消息未排队(rc == MQTT_ERR_QUEUE_SIZE)，它将引发ValueError，如果发布时出现错误，则可能是由于客户端未连接而引发RuntimeErro

如果主题为None，长度为零或无效(包含通配符)，如果qos不是0、1 或 2 之一，或者有效负载的长度大于 268435455 字节，则会引发ValueError

订阅、退订主题

subscribe

客户端订阅主题方法，可以通过3种方式订阅，mqttv5还有3种方式

client.subscribe(topic, qos, options, properties)

topic = '订阅的主题'
qos = '服务质量默认为0'
options:'订阅选项,mqtt v5 使用'
'''
订阅的选项必须是SubscribeOptions对象,位于 from paho.mqtt.subscribeoptions import SubscribeOptions
SubscribeOptions的构造参数如下

qos：与MQTT v3.1.1中相同。

noLocal：True或False。如果设置为True，则订阅者不会收到自己的发布。

retainAsPublished：True或False。如果设置为True，则收到的发布的retain标志将按发布者设置。

retainHandling：RETAIN_SEND_ON_SUBSCRIBE，RETAIN_SEND_IF_NEW_SUB或RETAIN_DO_NOT_SEND
                控制代理何时发送保留消息：
                RETAIN_SEND_ON_SUBSCRIBE：在任何成功的订阅请求上
                RETAIN_SEND_IF_NEW_SUB：仅在订阅请求是新的情况下
                RETAIN_DO_NOT_SEND：从不发送保留消息

'''



properties = '设置MQTT v5.0属性的properties实例，可选-如果不设置，则不发送任何属性'

字符串和整数订阅

client.subscribe('my/topic',2)
'''
topic:指定要订阅的订阅主题的字符串。
qos:订阅所需的服务质量级别，默认为0。
选项和属性:未使用
'''

字符串和整数的元组

client.subscribe(('my/topic',1))
'''
topic: (topic, qos)的元组
qos、选项和属性:未使用
'''

字符串和整数元组的列表

client.subscribe([('my/topcic',0),('my/topic2',1)])
'''
这允许在单个 SUBSCRIPTION 命令中进行多个主题订阅，这比使用多次调用subscribe()更有效
(topic, qos)的元组列表。topic 和 qos 都必须存在于所有元组中
qos、选项和属性:未使用
'''

字符串和订阅选项(仅MQTT v5.0)

from paho.mqtt.subscribeoptions import SubscribeOptions

client.subscribe('my/topic',options=SubscribeOptions(qos=2))
'''
topic:订阅主题。
qos:未使用。
选项:MQTT v5.0订阅选项。
properties:属性未设置
'''

字符串和订阅选项元组(仅MQTT v5.0)

from paho.mqtt.subscribeoptions import SubscribeOptions

client.subscribe(('my/topic'， SubscribeOptions(qos=1)))
'''
topic: (topic, SubscribeOptions)的元组。主题和订阅选项必须出现在元组中
qos、options、roperties未使用
'''

字符串和订阅选项元组列表(仅MQTT v5.0)

from paho.mqtt.subscribeoptions import SubscribeOptions
client.subscribe([('my/topic'， SubscribeOptions(qos=1)),('my/topic2'， SubscribeOptions(qos=2))])
'''
允许在单个SUBSCRIPTION中进行多个主题订阅命令，比使用多个调用更有效
topic:格式元组(topic, SubscribeOptions)的列表。主题和订阅选项必须出现在所有元组中。
qos和options:未使用。
properties:未设置
'''

订阅函数的返回

返回一个元组(result, mid)，其中result为MQTT_ERR_SUCCESS表示成功或(MQTT_ERR_NO_CONN, None)客户端当前未连接。

mid是订阅请求的消息 ID。mid 值可用于通过检查on_subscribe()回调中的 mid 参数(如果已定义)来跟踪订阅请求。

如果qos不是 0、1 或 2，或者主题为None或字符串长度为零，或者主题不是字符串、元组或列表，则引发ValueError

unsubscribe

取消订阅一个或多个主题

client.unsubscribe(topic,properties)
'''
topic:一个主题字符串或主题字符串列表，取消订阅的订阅主题
properties:默认值为None,mqttv5的订阅属性
'''

取消订阅函数的返回

返回一个元组(result, mid)，其中result是MQTT_ERR_SUCCESS表示成功，或者(MQTT_ERR_NO_CONN, None)如果客户端当前未连接。mid是取消订阅请求的消息 ID。mid 值可用于通过检查on_unsubscribe()回调中的 mid 参数(如果已定义)来跟踪取消订阅请求。

如果主题为None或字符串长度为零，或者不是字符串或列表，则引发ValueError 。

回调方法

回调是为响应事件而调用的函数,使用回调需要做两件事情,1.创建回调函数,2.将函数分配给回调

on_connect

客户端连接、重新连接broker后,连接的回调方法,当客户端收到broker的CONNACK响应的时候，会触发on_connect()回调

on_connect(client, userdata, flags, rc)
'''
client:触发回调的客户端实例
userdata:在Client()或者user_data_set()中设置的用户数据
flags:一个包含来自代理的响应标志的字典,仅使用 clean session 设置为 0。如果 clean session=0 的客户端重新连接到它之前连接过的代理，则此标志指示代理是否仍然具有客户端的会话信息。如果为 1，会话仍然存在
rc:连接结果,0：连接成功 1：连接被拒绝 - 协议版本不正确 2：连接被拒绝 - 客户端标识符无效 3：连接被拒绝 - 服务器不可用 4：连接被拒绝 - 用户名或密码错误 5：连接被拒绝 - 未授权 6-255：当前未使用

'''

使用示例

# 定义回调的方法  
def on_connect(client, userdata, flags, rc):
     print("Connected with result code: " + str(rc))

# 设置客户端的回调on_connect属性为定义的on_connect方法
client.on_connect = on_connect

on_disconnect

当客户端与代理断开时触发

on_disconnect(client, userdata, rc)
'''
client:触发的客户端实例
userdata:用户数据
rc:断开连接结果,如果是0，则回调是为了响应disconnect()调用而调用的。如果有任何其他值，则断开连接是意外的，例如可能由网络错误引起的
'''

使用示例

# 定义回调函数
def on_disconnect(client, userdata, rc):
  
    print("Unexpected disconnection %s" % rc)
    
# 设置客户端的回调on_disconnect属性为定义的on_disconnect方法
client.on_disconnect = on_disconnect

on_publish

当Clinet发送消息到broker,会触发on_publish()回调

on_publish(client, userdata, mid)
'''
mid:消息ID
'''

使用示例

def on_publish(client, userdata, mid):
  	print(mid,'message publish')
'''
触发该回调
对于 QoS 级别 1 和 2 的消息，意味着适当的握手已经完成
对于 QoS 0，这仅表示消息已离开客户端。mid变量匹配从相应的publish()调用返回的 mid 变量，以允许跟踪传出消息。
这个回调很重要，因为即使 publish() 调用返回成功，也并不总是意味着消息已发送
'''

on_message 和 message_callback_add

on_message,当 client 接收到已订阅的话题的消息并且与message_callback_add自定义主题筛选不匹配的时候，会调用 on_message() 回调函数

on_message(client, userdata, message)
'''
message:MQTTMessage类的实例。有topic, payload, qos, retain属性
'''

使用示例

def on_message(client, userdata, message):
    print("主题:" + msg.topic + " 消息:"+ str(message.payload.decode("utf-8")))
                                          
client.on_message = on_message

message_callback_add,用于处理特定订阅过滤器的传入消息，包括使用通配符，message_callback_add自定义是自定义的一个主题筛选过滤器,比如我用message_callback_add 筛选主题 A,有A的消息触发message_callback_add回调，否则触发on_message

message_callback_add(sub, callback)
'''
sub:此回调进行匹配的订阅筛选器
callback:要使用的回调
'''

使用示例

# 自定义处理action主题的回调
def action_message(client,userdata,message):
		print('action topic:'+message.topic)

client.message_callback_add('topic',action_message)

如果想要删除使用message_callback_add注册的特定回调

client.message_callback_remove('topic')

on_subscribe

当代理确认订阅时,将生成on_subscribe()回调

on_subscribe(client, userdata, mid, granted_qos)
'''
granted_qos:一个整数列表，给出代理为每个不同订阅请求授予的 QoS 级别
'''

使用示例

def on_subscribe(client, userdata, mid, granted_qos):
		print(f"On Subscribed: qos ={granted_qos}")
    
client.on_subscribe = on_subscribe

on_unsubscribe

当代理确认取消订阅时,生成on_unsubscribe()回调

def on_unsubscribe(client, userdata, mid):
  print('un unsubscribe ' + userdata)
  
client.on_unsubscribe = on_unsubscribe

on_log

当客户端有日志信息时调用

def on_log(client, userdata, level, buf):
    '''
    level:消息级别,MQTT_LOG_INFO、MQTT_LOG_NOTICE、MQTT_LOG_WARNING、MQTT_LOG_ERR和MQTT_LOG_DEBUG之一
    buf:日志消息
    '''
    # 可以与标准 Python 日志记录同时使用，可以通过enable_logger方法启用
    print(buf)
  
client.on_log = on_log

on_socket_open

当套接字打开时调用。使用它来向外部事件循环注册套接字以进行读取

def on_socket_open(client, userdata, sock):
		print(f'{socket} open')

client.on_socket_open = on_socket_open

on_socket_close

当套接字即将关闭时调用。使用它从外部事件循环中注销套接字以进行读取

def on_socket_close(client, userdata, sock):
				print(f'{socket} close')

client.on_socket_close = on_socket_close

on_socket_register_write

当对套接字的写操作失败时调用，因为它会阻塞，例如输出缓冲区已满。使用它来向外部事件循环注册套接字以进行写入

def on_socket_register_write(client, userdata, sock):
  	print(f'{socket} on_socket_register_write')
  
client.on_socket_register_write = on_socket_register_write

on_socket_unregister_write

当对套接字的写操作在先前失败后成功时调用。使用它从外部事件循环中注销套接字以进行写入。

def on_socket_unregister_write(client, userdata, sock):
  	print(f'{sock} on_socket_register_write')

client.on_socket_unregister_write = on_socket_unregister_write

外包事件循环支持

# 循环读取-当套接字准备好读取时调用
loop_read()

# 循环写入-当套接字准备好写入时调用
loop_write()

# 每隔几秒调用一次以处理消息重试和 ping
loop_misc()

# 返回客户端中使用的套接字对象，以允许与其他事件循环进行交互，对于基于选择的循环比较有用
socket()

# 如果有数据等待写入，则返回 true，以允许将客户端与其他事件循环连接起来，对于基于选择的循环比较有用
want_write()

'''
回调按以下顺序调优
1. on_socket_open 
2. on_socket_register_write - 零次或多次
3. on_socket_unregister_write - 零次或多次
4. on_socket_close
'''

全局辅助函数

客户端模块提供了一些全局辅助函数,对客户端行为提供一些行为操作

检查辅助函数

# 导入
import paho.mqtt.client as mqtt

topic_matches_sub

import paho.mqtt.client as mqtt
#检查主题与订阅是否匹配
mqtt.topic_matches_sub(sub, topic)

connack_string

# 返回与 CONNACK 结果关联的错误字符串，例如在连接回调中传入rc字段输出对应信息
mqtt.connack_string(connack_code)

error_string

# 返回与 Paho MQTT 错误号关联的错误字符串
mqtt.error_string(mqtt_errno)

发布辅助函数

允许以一次性方式直接发布消息。在有一条/多条消息要发布到代理的情况下很有用，然后断开连接而不需要其他任何东西，都对mqttv5协议支持，但是目前不允许在连接或者发布消息的时候设置属性

single

向代理发布一条消息，然后干净地断开连接

# 导包
from paho.mqtt.publish import single

# 参数
def single(topic, payload=None, qos=0, retain=False, hostname="localhost",
           port=1883, client_id="", keepalive=60, will=None, auth=None,
           tls=None, protocol=paho.MQTTv311, transport="tcp", proxy_args=None):
  
'''
topci:主题

payload:内容

qos:服务质量

retain:是否设置保留消息

hostname:/
port:/

client_id:要使用的 MQTT 客户端 ID。如果为“”或 None，Paho 库将自动生成一个客户端 ID

keepalive:心跳间隔
will:包含客户端遗嘱参数的字典，{'topic': 'topic',  'qos':1},主题是必须的,其他参数都是可选的

auth:包含客户端身份验证参数的字典,{'username':'liqi','password':'123456'}
		 默认为无，表示不使用身份验证,如果设置的话用户名是必需的，密码是可选的
		 
tls:包含客户端 TLS 配置参数的字典,{'ca_certs':'ca_certs'}
    ca_certs 是必需的，所有其他参数都是可选的,与tls_set方法参数一致
    
protocol:mqtt协议版本

transport：默认tcp,设置为“websockets”以通过 WebSockets 发送 MQTT

proxy_args:配置MQTT连接代理。启用对SOCKS或HTTP代理参数是字典类型
					 {'proxy_type':'代理类型，必选,socks.HTTP, socks.SOCKS4, or socks.SOCKS5',
					  'proxy_addr':'代理服务器的IP地址或DNS名称,必选',
					  'proxy_rdns':'bool,可选,是否应该执行代理查找,远程(True，默认)或本地(False)',
					  'proxy_username':'可选,SOCKS5代理的用户名，SOCKS4代理的用户id',
					  'proxy_password':'可选,SOCKS5代理密码'
					 }
					 该参数对应的使用方法表示在connect() or connect_async()之前使用
				 	   
'''

multiple

向代理发布多条消息，然后完全断开连接

# 导包
from paho.mqtt.publish import multiple

# 参数
def multiple(msgs, hostname="localhost", port=1883, client_id="", keepalive=60,
             will=None, auth=None, tls=None, protocol=paho.MQTTv311,
             transport="tcp", proxy_args=None):

'''
msgs:要发布的消息列表。每个消息要么是一个字典，要么是一个元组,如果是字典，则只有主题必须存在。默认值将用于任何缺少的参数如果是元组,参数按字段顺序对应
其他参数与single一致
'''

订阅辅助函数

允许直接订阅和处理消息,都包括对 MQTT v5.0 的支持，但目前不允许在连接或订阅时设置任何属性

simple

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

# 导包
from paho.mqtt.subscribe import simple

# 参数
def simple(topics, qos=0, msg_count=1, retained=True, hostname="localhost",
           port=1883, client_id="", keepalive=60, will=None, auth=None,
           tls=None, protocol=paho.MQTTv311, transport="tcp",
           clean_session=True, proxy_args=None):


'''
topics:是客户端将订阅的主题字符串。如果订阅多个主题，可以是字符串或字符串列表
msg_count:消息计数,从代理检索的消息数。默认为 1。如果为 1，将返回单个 MQTTMessage 对象。如果 >1，将返回 MQTTMessages 列表
retained:设置为 True 以保留消息，设置为 False 以忽略设置了保留标志的消息
clean_session:会话设置

其他参数参考single

'''

使用示例

msg = simple("my/test")
rint("%s %s" % (msg.topic, msg.payload))

callback

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

# 导包
from paho.mqtt.subscribe import callback

# 参数
def callback(callback, topics, qos=0, userdata=None, hostname="localhost",
             port=1883, client_id="", keepalive=60, will=None, auth=None,
             tls=None, protocol=paho.MQTTv311, transport="tcp",
             clean_session=True, proxy_args=None):
  
'''
callback:回调,定义一个回调,将用于收到的每条消息，形式与on_message类似
userdata:用户数据,将在收到消息时传递给 on_message 回调
其他参数参考simple
'''

使用示例

# 定义一个回调
def on_message_print(client, userdata, message)
    print("%s %s" % (message.topic, message.payload))

callback(callback=on_message_print, topics="my/test")