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)/“物联网”连接协议。它被设计为一种极其轻量级的发布/订阅消息传输，对于需要小代码占用和/或网络带宽非常昂贵的远程连接非常有用。
安装

1. 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主题树并打印出结果消息：

1. # -*- coding:utf-8 -*-
3. import paho.mqtt.client as mqtt
5. def on_connect(client, userdata, flags, reason_code, properties):
6.     '''客户端从服务器接收到 CONNACK 响应时的回调'''
8.     print(f"Connected with result code {reason_code}")  # 成功连接时 reason_code 值为 Success
10.     # 在on_connect()中执行订阅操作，意味着如果应用失去连接并且重新连接后，订阅将被续订。
11.     if reason_code == 'Success':
12.         client.subscribe('$SYS/#')
14. def on_disconnect(client, userdata, flags, reason_code, properties):
15.     print(f'Disconnected with result code {reason_code}')
18. def on_message(client, userdata, msg):
19.     '''从服务器收到 PUBLISH 消息时的回调。'''
20.     print(msg.topic + ' ' + str(msg.payload)) # 输出值形如 $SYS/broker/version b'mosquitto version 2.0.18'
22. mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
23. mqttc.on_connect = on_connect
24. mqttc.on_disconnect = on_disconnect
25. mqttc.on_message = on_message
27. # client.username_pw_set('testacc', 'test1234') # 设置访问账号和密码
29. mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)
31. # 阻塞调用，处理网络流量、分派回调和处理重新连接
32. # 有其它提供线程接口和手动接口的loop*()函数可用
33. 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。
      
    
    
  
  
• 1. 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类一般使用流程如下：

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

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

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

1. mqttc.loop_start()
3. while True:
4.     temperature = sensor.blocking_read()
5.     mqttc.publish("paho/temperature", temperature)
7. mqttc.loop_stop()

复制代码

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

1. mqttc.loop_forever(retry_first_connection=False)

复制代码

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

1. run = True
2. while run:
3.     rc = mqttc.loop(timeout=1.0)
4.     if rc != 0:
5.         # 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)的回调。详见下文。
  

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

1. # -*- coding:utf-8 -*-
3. import paho.mqtt.client as mqtt
5. def on_subscribe(client, userdata, mid, reason_code_list, properties):
6.     # 由于我们只订阅了一个信道，reason_code_list只包含一个条目
7.     # print(reason_code_list) #输出： [ReasonCode(Suback, 'Granted QoS 0')]
8.     if reason_code_list[0].is_failure:
9.         print(f"Broker rejected you subscription: {reason_code_list[0]}")
10.     else:
11.         print(f"Broker granted the following QoS: {reason_code_list[0].value}")
13. def on_unsubscribe(client, userdata, mid, reason_code_list, properties):
14.     #注意，reason_code_list仅存在于MQTTv5中，在MQTTv3中，它将始终为空
15.     if len(reason_code_list) == 0 or not reason_code_list[0].is_failure:
16.         print("unsubscribe succeeded (if SUBACK is received in MQTTv3 it success)")
17.     else:
18.         print(f"Broker replied with failure: {reason_code_list[0]}")
19.     client.disconnect()
21. def on_message(client, userdata, message):
22.     # userdata是我们选择提供的数据结构，这里为一个列表(通过下方的 mqttc.user_data_set([])设置，该函数参数即为userdata参数值
23.     userdata.append(message.payload)
24.     # 假设只想处理10条消息
25.     if len(userdata) >= 10:
26.         client.unsubscribe("$SYS/#")
28. def on_connect(client, userdata, flags, reason_code, properties):
29.     if reason_code.is_failure:
30.         print(f"Failed to connect: {reason_code}. loop_forever() will retry connection")
31.     else:
32.         # 应该始终在 on_connect 回调中订阅以确保在重新连接时订阅依旧存在。
33.         client.subscribe("$SYS/#")
35. mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
36. mqttc.on_connect = on_connect
37. mqttc.on_message = on_message
38. mqttc.on_subscribe = on_subscribe
39. mqttc.on_unsubscribe = on_unsubscribe
41. mqttc.user_data_set([]) # 设置 userdata
42. mqttc.connect("mqtt.eclipseprojects.io")
43. mqttc.loop_forever() # 当调用client.disconnect()后继续执行以下代码
45. print(f"Received the following message: {mqttc.user_data_get()}")

复制代码

发布示例

1. # -*- coding:utf-8 -*-
3. import time
4. import paho.mqtt.client as mqtt
6. def on_publish(client, userdata, mid, reason_code, properties):
7.     '''reason_code和properties将仅出现在MQTTv5中。在MQTTv3中始终未设置
8.     使用不存在`uncaked_publish`中的`mid`调用`on_publish()`。这是由于不可避免的竞争情形：
9.     * publish() 返回已发送消息的mid。
10.     * 主线程将publish()返回的mid添加到uncaked_publish中
11.     * loop_start线程调用on_publish()
12.     虽然不太可能（因为on_publish()将在网络往返后调用），但是这是一种可能发生的竞争情形
13.     避免竞争情形的最佳解决方案是使用publish()中的msg_info。还可以尝试使用已确认的mid列表，而不是从待处理列表中删除
14.     但是请记住，mid可以重复使用！
15.     reason_code和properties将仅出现在MQTTv5中。在MQTTv3中始终未设置
16.     '''
17.     try:
18.         userdata.remove(mid)
19.     except KeyError:
20.         print("on_publish() is called with a mid not present in unacked_publish")
21.         print("This is due to an unavoidable race-condition:")
22.         print("* publish() return the mid of the message sent.")
23.         print("* mid from publish() is added to unacked_publish by the main thread")
24.         print("* on_publish() is called by the loop_start thread")
25.         print("While unlikely (because on_publish() will be called after a network round-trip),")
26.         print(" this is a race-condition that COULD happen")
27.         print("")
28.         print("The best solution to avoid race-condition is using the msg_info from publish()")
29.         print("We could also try using a list of acknowledged mid rather than removing from pending list,")
30.         print("but remember that mid could be re-used !")
32. unacked_publish = set()
33. mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
34. mqttc.on_publish = on_publish
36. mqttc.user_data_set(unacked_publish)
37. mqttc.connect("mqtt.eclipseprojects.io")
38. mqttc.loop_start()
40. # 应用生产一些消息
41. msg_info = mqttc.publish("paho/test/topic", "my message", qos=1)
42. unacked_publish.add(msg_info.mid)
44. msg_info2 = mqttc.publish("paho/test/topic", "my message2", qos=1)
45. unacked_publish.add(msg_info2.mid)
47. # 等待所有消息被发布
48. while len(unacked_publish):
49.     time.sleep(0.1)
51. # 由于上述描述的竞争状态, 以下等待所有消息发布完成的方式更安全
52. msg_info.wait_for_publish()
53. msg_info2.wait_for_publish()
55. mqttc.disconnect()
56. 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()。可以提供自定义记录器或使用默认记录器
示例:

1. import logging
2. import paho.mqtt.client as mqtt
4. logging.basicConfig(level=logging.DEBUG)
6. mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
7. mqttc.enable_logger()
9. mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)
10. mqttc.loop_start()
12. # Do additional action needed, publish, subscribe, ...
13. [...]

复制代码

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

1. import paho.mqtt.client as mqtt
3. def on_log(client, userdata, paho_log_level, messages):
4.     if paho_log_level == mqtt.LogLevel.MQTT_LOG_ERR:
5.         print(message)
7. mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
8. mqttc.on_log = on_log
10. mqttc.connect("mqtt.eclipseprojects.io", 1883, 60)
11. mqttc.loop_start()
13. # Do additional action needed, publish, subscribe, ...
14. [...]

复制代码

Paho日志级别和标准日志级别的对应关系如下：
PahologgingMQTT_LOG_ERRlogging.ERRORMQTT_LOG_WARNINGlogging.WARNINGMQTT_LOG_NOTICElogging.INFO (no direct equivalent)MQTT_LOG_INFOlogging.INFOMQTT_LOG_DEBUGlogging.DEBUG外部事件循环支持

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

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

用伪代码表示如下:

1. while run:
2.     if need_read:
3.         mqttc.loop_read()
4.     if need_write:
5.         mqttc.loop_write()
6.     mqttc.loop_misc()
8.     if not need_read and not need_write:
9.         # But don't wait more than few seconds, loop_misc() need to be called regularly
10.         wait_for_change_in_need_read_or_write()
11.     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

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

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

复制代码

Multiple

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

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

复制代码

订阅

此模块提供了两个辅助函数simple()和callback()，以允许直接订阅和处理消息。
这两个函数都支持MQTT v5.0，但目前不允许在连接或发送消息时设置任何属性。
Simple

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

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

复制代码

使用回调(Callback)

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

1. import paho.mqtt.subscribe as subscribe
3. def on_message_print(client, userdata, message):
4.     print("%s %s" % (message.topic, message.payload))
5.     userdata["message_count"] += 1
6.     if userdata["message_count"] >= 5:
7.         # it's possible to stop the program by disconnecting
8.         client.disconnect()
10. 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

来源:https://www.cnblogs.com/shouke/p/18417803
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！