ITADN
peterhinch/micropython-mqtt
代码 动态 Issues58 合并请求9 版本
peterhinch/micropython-mqtt · 文件 下载 ZIP
文件最后提交记录最后更新时间
README.md
以下内容由 AI 翻译，如有问题请点此提交 issue 反馈

MicroPython 异步 MQTT

MQTT 是一种易于使用的网络协议，专为 IOT（物联网）应用而设计。它非常适合用于控制硬件设备， 以及通过本地网络或互联网读取传感器数据。

它是多个客户端之间通信的一种方式。单个服务器，也称为 broker，负责管理网络。客户端可以包括 ESP8266、ESP32 和 Pyboard D 模块以及其他联网计算机。典型的服务器硬件是 Raspberry Pi 或其他小型 Linux 机器，可以 24/7 持续运行。一个有效的 PC 服务器是 mosquitto。公共 broker 也存在。

MQTT 数据包使用发布/订阅模型在客户端之间传递。它们由一个主题和一个消息字符串组成。客户端订阅 某个主题后，将接收任何客户端在该主题下发布的所有数据包。

该协议支持三种“服务质量”（qos）级别。级别 0 不提供任何保证。级别 1 确保数据包被传达给接收方， 但可能会发生重复。级别 2 避免重复；官方驱动程序或此模块均不支持该级别。重复可以在应用层面轻松处理。

主 README

警告：固件 >= V1.22.0

V1.22.0 包含了更改后的 IDF 版本 5.0.4。如果在 ESPx 上升级固件， MQTT 包也应为最新版本，否则可能无法从故障中恢复。

1. 目录

  1. Contents
    1.1 Rationale
    1.2 Overview
    1.3 Project Status
    1.4 ESP8266 limitations
    1.5 ESP32 Issues
    1.6 Pyboard D
    1.7 Arduino Nano RP2040 Connect
    1.8 RP2 Pico W
    1.9 Limitations Please read this.
    1.10 MQTTv5 Which version should you use?
  2. Getting started
    2.1 Program files Quick installation and setup.
    2.2 Installation on ESP8266
    2.3 Example Usage Using the event interface.
    2.4 Usage with callbacks
  3. MQTTClient class
    3.1 Constructor Describes the MQTT configuration dictionary.
    3.2 Methods
         3.2.1 connect
         3.2.2 publish
         3.2.3 subscribe
         3.2.4 unsubscribe
         3.2.5 isconnected
         3.2.6 disconnect
         3.2.7 close
         3.2.8 broker_up
         3.2.9 wan_ok
         3.2.10 dprint
    3.3 Class Variables
    3.4 Module Attribute
    3.5 Event based interface
    3.6 MQTTv5 Support
         3.6.1 Configuration and Migration from MQTTv3.1.1
         3.6.2 MQTTv5 Properties
         3.6.3 Unsupported Features
  4. Notes
    4.1 Connectivity
    4.2 Client publications with qos == 1
    4.3 Client subscriptions with qos == 1
    4.4 Application Design
         4.4.1 Publication Timeouts
         4.4.2 Behaviour on power up
         4.4.3 Optimisations RAM use, large incoming messages.
    4.5 Alternative design approach Continue the MQTT paradigm into the application.
  5. Non standard applications Usage in specialist and micropower applications.
    5.1 deepsleep
    5.2 lightsleep and disconnect
    5.3 Ultra low power consumption For ESP8266 and ESP32.
  6. References
  7. Connect Error Codes
  8. Hive MQ A secure, free, broker.
  9. The ssl_params dictionary Plus user notes on SSL/TLS.

1.1 设计理由

官方的“健壮”MQTT 客户端存在以下局限性。

  1. 在临时 WiFi 中断后，它无法可靠地恢复运行。

  2. 它使用阻塞套接字，在访问缓慢的 broker 时可能导致执行暂停任意 时间。在 qos == 1 发布的情况下，如果等待一个永远不会到达的发布确认， 它也可能永久阻塞；如果序列在此时发生中断，这种情况可能会在 WiFi 网络上 出现。

  3. 这种阻塞行为意味着与异步应用的兼容性有限，因为在阻塞期间挂起的协程 不会被调度。

  4. 它对 qos == 1 的支持是不完整的。它不支持在发布确认丢失的情况下进行 重传。这种情况可能会在 WiFi 网络上发生，尤其是在接近范围极限或存在干扰 时。

  5. 其不完整的 qos == 1 支持以及无法在 WiFi 中断后可靠恢复的能力，限制了 可用的 WiFi 范围。为了实现可靠运行，客户端必须位于接入点 (AP) 的良好 范围内。

  6. 作为一个同步解决方案，它没有机制来支持 MQTT 的“keepalive” 机制。这导致“last will”系统无法正常工作。它还使得仅订阅客户端变得 有问题：broker 没有“知道”客户端是否仍然连接的手段。

本模块旨在解决这些问题，代价是显著增加代码体积。 它已在以下平台上进行了测试。

  1. ESP8266
  2. ESP32、ESP32-S2 和 ESP32-S3
  3. Pyboard D
  4. Arduino Nano Connect
  5. Raspberry Pi Pico W

该驱动的主要特性是：

  1. 支持使用 uasyncio 的应用程序的非阻塞操作。
  2. 自动从 WiFi 和 broker 故障中恢复。
  3. 支持重传的真正 qos == 1 操作。
  4. 由于对较差连接性的容忍度，改善了 WiFi 覆盖范围。

其缺点是代码体积增加，这在 ESP8266 上是一个问题。 作为冻结字节码运行时，它在 ESP8266 上约占 50% 的 RAM。在 ESP32 和 Pyboard D 上，它可以作为标准 Python 模块运行。

1.2 概述

本模块提供了一个“弹性”非阻塞 MQTT 驱动。在此上下文中， “弹性”意味着在存在较差 WiFi 连接性和中断的情况下能够可靠运行。显然，在中断或 broker 故障期间 通信是不可能的，但当连接恢复时，驱动会透明地恢复。

在 WiFi 覆盖范围的极限附近，由于重传和重连，可能会产生通信延迟，但非阻塞行为 和 qos == 1 完整性得以保持。

它支持 qos 级别 0 和 1。对于 qos == 1 的数据包， 在数据包成功传输之前会发生重传。 如果 WiFi 失效（例如设备移出 AP 的范围）， 执行发布的协程将暂停，直到连接恢复。

该驱动程序需要 asyncio 库，并旨在用于使用 该库的应用程序。它使用非阻塞套接字，不会阻塞调度器。其 设计基于官方的 umqtt 库，但为了增强韧性和 异步操作进行了大幅修改。

它主要旨在用于打开与 MQTT 代理的连接 并旨在无限期保持该连接的应用程序。关闭并 重新打开连接的应用程序（例如出于省电目的）受到 非标准应用程序 中详述的限制。

硬件支持：Pyboard D、ESP8266、ESP32、ESP32-S3、ESP32-S2、Pico W 以及 Arduino Nano RP2040 Connect。
固件支持：官方 MicroPython 固件 V1.19 或更高版本。
代理支持：推荐 Mosquitto，因其具有出色的 MQTT 合规性。
协议：该模块支持 MQTT 修订版 3.1.1 的子集。

1.3 项目状态

初始开发由 Peter Hinch 完成。感谢 Kevin Köck 提供并测试了多项错误修复和增强功能。也感谢其他 贡献者，其中一些在下文中提及。

请注意，在 1.21 之前的固件中，asyncio 的名称为 uasyncio。

2025 年 3 月 7 日 V0.8.3 修复取消订阅 bug。修复大变量字节整数的解码。 2024 年 10 月 24 日 V0.8.2 Socket 读取使用预分配缓冲区以提升性能。 2024 年 8 月 18 日 V0.8.1 重构为 Python 包。修复 V5 支持中的 bug。 2024 年 8 月 9 日 V0.8.0 由 Bob Veringa 贡献的部分 MQTTv5 支持。 2024 年 2 月 15 日 V0.7.2 使其符合固件 V1.22.0 及更高版本。 2022 年 11 月 12 日 V0.7.0 提供替代的无回调 Event 接口。 2022 年 11 月 2 日 将 config.py 重命名为 mqtt_local.py，改进文档。 2022 年 8 月 8 日 V0.6.6 支持取消订阅（感谢 Kevin Köck 的 fork）。 2022 年 7 月 11 日 V0.6.5 支持 RP2 Pico W 2022 年 7 月 5 日 V0.6.4 实现来自 Bob Veringa 的增强功能。修复任务在短暂中断时可能无法停止的 bug。订阅回调现在接收 bytearrays 而不是 bytes 对象。 2022 年 6 月 10 日 移除低功耗演示，因为它需要 asyncio 的过时版本。 改进了对 clean_init 的处理（issue #40）。 2022 年 5 月 21 日 SSL/TLS ESP8266 支持由 @SooOverpowered 贡献：参见 tls8266.py。 2022 年 4 月 22 日 添加对 Arduino Nano RP2040 Connect 的支持。参见下文说明。 2021 年 8 月 2 日 ESP32 上的 SSL/TLS 现已确认可用。 参考。

1.4 ESP8266 限制

该模块过大，无法在 ESP8266 上编译，应预先编译或 最好冻结为字节码。在参考板上冻结 mqtt_as 后， 演示脚本 range_ex 在运行时报告有 27.4K 的可用 RAM。该代码 禁用了自动睡眠：这以减少重连次数为代价，增加了功耗。

有关 Sonoff Basic R3 的说明可在 此处 找到。

1.5 ESP32 问题

固件现在必须是上述官方固件。Loboris 移植版 已被其作者放弃，不再受支持。

1.6 Pyboard D

该库已在 Pyboard D SF2W 和 SF6W 上成功测试。在 测试中，它已记录连续运行八周且处理近 100 万条 消息，未发生故障或数据丢失。

1.7 Arduino Nano RP2040 Connect

NINA 固件必须保持最新，否则 MicroPython 会产生错误消息。 参见 此文档。 读取 RSSI 似乎会破坏 WiFi 连接，因此应避免 - range_ex.py 演示在此平台上禁用了此功能。

1.8 RP2 Pico W

mqtt_as 代码应为 V0.6.5 或更高版本，以避免从 故障中恢复速度极慢的问题。

1.9 局限性

MQTT 3.1 协议支持极长的消息。在微控制器上， 消息长度受可用 RAM 限制。实际限制取决于 平台和用户代码，但最好基于最大约 1KiB 进行设计。

避免对性能抱有不切实际的期望：延迟可能相当显著， 尤其是在使用 TLS 连接连接到位于互联网上的代理时。 使用非加密连接连接到本地代理时，使用一个 MicroPython 客户端来控制另一个客户端是可行的。我没有测量延迟，但我估计 约为 ~100ms。

某些平台——尤其是 ESP32——在处理严重错误时 （例如 WiFi 凭据不正确）并不友好。初始连接只会在一分钟超时后 失败。其他平台允许立即退出。

1.10 MQTTv5

MQTTv5 支持的新增不会影响现有应用程序，这些应用程序 将保持不变地运行。预计大多数微控制器用户将继续使用 MQTT V3.1.1。使用 MQTTv5 会消耗额外的 RAM（约 3KiB），并且需要一些 协议知识。有关更多详细信息，请参阅 MQTTv5 支持 。

目录

2. 入门

2.1 程序文件

该库被配置为 Python 包，并安装到 mqtt_as 目录中。这通常位于 Python 路径上。然后演示脚本 使用以下方式运行（例如）：

>>> import mqtt_as.range

用户配置的 mqtt_local.py（见下文）位于 Python 路径上。

必需文件

  1. __init__.py 主模块。
  2. mqtt_v5_properties.py 仅在使用 MQTTv5 时需要。

演示脚本所需

  1. mqtt_local.py 保存本地配置详细信息，例如 WiFi 凭据。 放置在 Python 路径上（通常是 / 或 /lib/）。

测试/演示脚本

前两个示例展示了事件接口。其余的则使用回调。

  1. range.py 用于 WiFi 范围测试。良好的通用演示。
  2. range_ex.py 同上，但还发布 RSSI 和空闲 RAM。请参阅代码 注释以了解 Pico W 和 Arduino nano connect 上的限制。
  3. clean.py 使用 MQTT Clean Session 的测试/演示程序。
  4. unclean.py 带有 MQTT Clean Session False 的测试/演示程序。
  5. main.py 自动启动应用程序的示例。
  6. tls.py 与公共 broker 建立 SSL/TLS 连接的演示。此程序运行在 Pyboard D 上。每 20 秒发布一次并订阅同一主题。与 此公共 broker 的连接虽然经过加密，但不安全，因为任何人 都可以订阅。
  7. tls8266.py ESP8266 的 SSL/TLS 连接。演示如何使用密钥和 证书。出于显而易见的原因，运行前需要进行编辑。
  8. sub_unsub.py 主题为 sub_topic 的消息控制对 另一主题的订阅。

MQTTv5 的测试脚本：

  1. basic.py MQTTv5 下用户属性的演示。

Bash 脚本（可在 PC 上运行以定期发布）：

  1. pubtest 演示使用 Mosquitto 进行发布的 Bash 脚本。
  2. pubtest_v5 演示各种发布属性的 Bash 脚本。

快速安装

ESP8266：请阅读 在 ESP8266 上安装。在其他 平台上，主模块、演示 1 到 3 以及示例 mqtt_local_example.py 可以从连接的 PC 上通过以下命令安装：

$ mpremote mip install github:peterhinch/micropython-mqtt

文件 mqtt_local_example.py 应针对本地 WiFi 认证进行编辑 并重命名为 mqtt_local.py。

对于 MQTTv5，可以通过以下方式添加演示

$ mpremote mip install github:peterhinch/micropython-mqtt/mqtt_as/v5

另一种方法是在连接 WiFi 的情况下在 REPL 中使用 mip：

>>> import mip
>>> mip.install("github:peterhinch/micropython-mqtt")

Bash 脚本 pubtest 和 pubtest_v5 应复制到 PC 上。

配置

MQTT 客户端使用字典进行配置。在 MQTTClient 类 中定义了一个名为 config 的实例，并填充了常见的默认值。用户可以以任何方式填充此内容。 测试脚本中采用的方法如下。主 __init__.py 模块使用典型默认值实例化 config。然后 mqtt_local.py 添加 所有节点共用的本地设置，例如 WiFi 凭据和 broker 详细信息。 最后，应用程序添加应用程序特定的设置，例如订阅。

在典型项目中，将编辑 mqtt_local.py 然后部署到所有节点。

ESP8266 内部存储 WiFi 凭据：如果 ESP8266 在运行之前已连接到 LAN，则无需显式指定这些凭据。在其他 平台上，或者为了具备在未连接过的 ESP8266 上运行的能力，应编辑 mqtt_local.py 以提供它们。这是 一个跨平台示例文件：

from mqtt_as import config

config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

# Required on Pyboard D and ESP32. On ESP8266 these may be omitted (see above).
config['ssid'] = 'my_WiFi_SSID'
config['wifi_pw'] = 'my_password'
内容

2.2 在 ESP8266 上安装

该模块太大，无法在 ESP8266 上编译。它必须交叉 编译，或者（最好）构建为冻结字节码：将 __init__.py 复制到 源树中的 esp8266/modules，构建并部署。将 mqtt_local.py 复制到 文件系统，以便于进行更改。

在其他平台上，只需将 Python 源代码复制到文件系统（至少包括上述第 1 项和第 2 项）。

如果应用程序需要在开机时自动运行，可能需要在 main.py 中添加一个短暂的 延迟：

import time
time.sleep(5)  # Could probably be shorter
import range  # Your application

这取决于平台，并给硬件留出初始化时间。

2.3 使用示例

该库提供了两种处理事件（例如消息到达）的替代方式。一种使用传统的回调。以下使用 Event 实例 和异步迭代器。如果 PC 客户端发布主题为 foo_topic 的消息，则会打印主题和消息。代码定期在主题 result 下发布递增的计数。

from mqtt_as import MQTTClient, config
import asyncio

# Local configuration
config['ssid'] = 'your_network_name'  # Optional on ESP8266
config['wifi_pw'] = 'your_password'
config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

async def messages(client):  # Respond to incoming messages
    # If MQTT V5is used this would read
    # async for topic, msg, retained, properties in client.queue:
    async for topic, msg, retained in client.queue:
        print(topic.decode(), msg.decode(), retained)

async def up(client):  # Respond to connectivity being (re)established
    while True:
        await client.up.wait()  # Wait on an Event
        client.up.clear()
        await client.subscribe('foo_topic', 1)  # renew subscriptions

async def main(client):
    await client.connect()
    for coroutine in (up, messages):
        asyncio.create_task(coroutine(client))
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

config["queue_len"] = 1  # Use event interface with default queue size
MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)
try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

可以通过在一个终端中运行 pubtest，在另一个终端中运行 mosquitto_sub -h 192.168.0.10 -t result 来测试代码（将 IP 地址更改为与您的 broker 匹配）。

2.4 使用回调

替代的基于回调的接口可以按以下方式运行：

from mqtt_as import MQTTClient, config
import asyncio

# Local configuration
config['ssid'] = 'your_network_name'  # Optional on ESP8266
config['wifi_pw'] = 'your_password'
config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

def callback(topic, msg, retained, properties=None):  # MQTT V5 passes properties
    print((topic.decode(), msg.decode(), retained))

async def conn_han(client):
    await client.subscribe('foo_topic', 1)

async def main(client):
    await client.connect()
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

config['subs_cb'] = callback
config['connect_coro'] = conn_han

MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)
try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

如上所述，测试是在一个终端中运行 pubtest，在另一个终端中运行 mosquitto_sub -h 192.168.0.10 -t result（将 IP 地址更改为与您的 broker 匹配）。

目录

3. MQTTClient 类

该模块提供了一个类：MQTTClient。

3.1 构造函数

它接受一个字典作为参数。默认值为 mqtt_as.config，其中 填充了下面列出的默认值。典型的应用程序会导入此字典 并根据需要修改选定的条目。条目如下（默认 值显示在 [] 中）：

WiFi 凭据

这些参数对于 ESP8266 以外的平台是必需的，而在 ESP8266 上是可选的。如果 ESP8266 之前已连接到所需的 LAN，芯片可以自动重新连接。如果提供了凭据， 没有存储值或存储值与任何可用网络不匹配的 ESP8266 将尝试连接到指定的 LAN。

'ssid' [None]
'wifi_pw' [None]

MQTT 参数

'client_id' [自动生成的唯一 ID] 必须是 bytes 实例。
'server' [None] Broker IP 地址（必填）。
'port' [0] 0 表示默认端口（SSL 为 1883 或 8883）。
'user' [''] MQTT 凭据（如需要）。
'password' [''] 如果提供了密码，则必须同时存在用户。
'keepalive' [60] Broker 认为客户端已断开前的周期（秒）。
'ping_interval' [0] Broker ping 之间的周期（秒）。0 == 使用默认值。
'ssl' [False] 如果 True 则使用 SSL。
'ssl_params' [{}] 见下文。
'response_time' [10] 预期服务器响应的时间（秒）。见下方说明。
'clean_init' [True] 在初始连接时清理会话状态。（如果使用 MQTT V5 则忽略）。
'clean' [True] 在重新连接时清理会话状态。（在 MQTT V5 中称为 Clean Start）。
'max_repubs' [4] 在尝试重新连接之前的最大重新发布次数。
'will' : [None] 定义最后遗嘱的列表或元组（见下文）。

接口定义

'queue_len' [0] 如果传入的值 > 0，则启用基于事件的接口。 这将用消息队列和 Event 实例替换下面定义的回调。参见 第 3.5 节。

基于回调的接口

此接口是可选的。它保留以兼容现有 代码。在新设计中，请考虑使用基于事件的接口，该接口 用更符合 asyncio 的方式替换回调。

'subs_cb' [一个空 lambda 函数] 订阅回调。当收到主题 匹配订阅的消息时运行。回调必须接受三个或 四个参数，topic、message、retained 和 properties=None。前两个 是 bytes 实例，retained 是 bool，True 如果消息是 保留消息。properties 是一个字典（或 None），如果使用了 MQTT V5。
'wifi_coro' [一个空协程] 一个协程。定义在网络 状态改变时运行的任务。该协程接收单个 bool 参数，即网络状态。
'connect_coro' [一个空协程] 一个协程。定义在与 broker 建立连接后运行的任务。这通常用于 注册和更新订阅。该协程接收单个参数，即 客户端实例。

MQTT V5 扩展

参见 MQTTv5 支持
'mqttv5' [False]
'mqttv5_con_props' [None]

备注

response_time 条目按以下方式工作。如果读或写操作超时， 连接将被视为已断开，重连过程随即开始。如果在此 期间 qos==1 发布未得到确认，将发生重新发布。对于慢速互联网连接，可能需要延长该时间。

will 条目定义了一种发布，当代理确定连接已超时时， 它将发出该发布。这是一个元组或列表，由 [topic (string), msg (string), retain (bool), qos (0 or 1)] 组成。如果提供了 arg， 则所有元素均为必填项。

Clean sessions：如果设置了 clean，则在服务中断期间来自服务器的消息 将丢失，无论其 qos 级别如何。

如果 clean 为 False，则服务器发送的 qos == 1 的消息将在 连接恢复时收到。这是标准的 MQTT 行为（MQTT 规范 第 3.1.2.4 节）。如果中断时间较长，这可能导致大量 积压。在 ESP8266 上，这可能导致 Espressif WiFi 堆栈中的缓冲区溢出，从而出现 LmacRxBlk:1 错误。 参见此文档。

clean_init 通常应为 True。如果 False，系统将在首次连接时尝试 恢复先前的会话。这可能导致收到大量 qos==1 消息的积压，例如当客户端长时间停止服务时。这可能会产生上述后果。 参见 MQTT 规范 3.1.2.4。这将在下文 第 4.4.2 节 上电时的行为] 中进一步描述。

SSL/TLS

填充 ssl_params 字典在某种程度上是一门玄学。某些站点 需要证书：请参阅 此帖子 以了解如何指定这些证书的详细信息。请参阅 Hive MQ 以 了解连接到安全、免费代理服务的详细信息。这可能为 连接到其他 TLS 代理提供提示。请参阅 下方的 The ssl_params dictionary。

Contents

3.2 方法

关于数据类型的说明。只要所有 字符的序数值 <= 127（Unicode 单字节字符），消息和主题可以是字符串。 否则应使用 encode 方法将它们转换为 bytes 对象。

3.2.1 connect

异步。

仅限关键字参数：

  • quick=False 设置 quick=True 在某些电池应用中可节省电量。 它通过在（重新）连接时跳过对 WiFi 完整性的检查来实现这一点。此检查 旨在用于可能在信号强度较差的条件下尝试重新连接的移动客户端。在信号强度良好的条件下，可以 跳过此检查。
    请参阅 Non standard applications。

连接到指定的代理。应用程序应在 启动时调用一次 connect。如果失败（由于 WiFi 或代理不可用），将 引发 OSError：请参阅 Connect Error Codes。中断后的 后续重连会自动处理。

3.2.2 publish

异步。

如果连接正常，协程将立即完成，否则它将暂停 直到 WiFi/代理可访问。 Section 4.2 描述了 qos == 1 操作。

Args:

  1. topic 一个 bytes 或 bytearray 对象。或如上所述的 ASCII 字符串。
  2. msg 一个 bytes 或 bytearray 对象。
  3. retain=False 布尔值。
  4. qos=0 整数。
  5. properties=None 参见 MQTTv5 Support。

3.2.3 subscribe

异步。

应在 connect 协程中创建订阅，以确保在故障后重新建立。

该协程将暂停，直到从 broker 接收到 SUBACK，如有必要，会重新连接到故障网络。

Args:

  1. topic 一个 bytes 或 bytearray 对象。或如上所述的 ASCII 字符串。
  2. qos=0 整数。

可以订阅多个主题，但只能有一个订阅回调。

3.2.4 unsubscribe

异步。

该协程将暂停，直到从 broker 接收到 UNSUBACK，如有必要，会重新连接到故障网络。

Arg:

  1. topic 一个 bytes 或 bytearray 对象。或如上所述的 ASCII 字符串。

如果不存在与传入的主题名称匹配的订阅，该方法将正常完成。这符合 MQTT 规范 3.10.4 Response。

3.2.5 isconnected

同步。无参数。

如果连接正常则返回 True，否则返回 False 并安排重连尝试。

3.2.6 disconnect

异步。无参数。

向 broker 发送 DISCONNECT 数据包，关闭 socket。断开连接会抑制 Will（MQTT 规范 3.1.2.5）。这可以在断电或深度睡眠之前执行。关于此方法的使用限制，请参见 lightsleep and disconnect。

3.2.7 close

同步。无参数。

关闭 WiFi 接口并关闭套接字。其主要用途是在开发阶段，防止应用程序抛出异常或通过 ctrl-C 终止时导致 ESP8266 LmacRxBlk:1 故障（参见 示例用法。

3.2.8 broker_up

异步。无参数。

除非在过去一秒内接收到了数据，否则它会发送一个 MQTT ping 并等待响应。如果超时（超过 response_time）且没有响应，则 返回 False，否则返回 True。

3.2.9 wan_ok

异步。

如果互联网连接可用，则返回 True，否则返回 False。它首先 检查当前的 WiFi 和 broker 连接。如果存在，则向 '8.8.8.8' 发送 DNS 查询 并检查是否有有效响应。

有一个参数 packet，它是一个 bytes 对象，即 DNS 查询。 默认对象查询 Google DNS 服务器。

请注意，这只是一个便捷方法。它不被 客户端代码使用，其使用完全是可选的。

3.2.10 dprint

如果类变量 DEBUG 为 true，则通过 dprint 输出调试消息。 此方法可以在子类中重新定义，例如将调试输出记录到 文件中。该方法接受任意数量的位置参数，如 print 所述。

3.3 类变量

  1. DEBUG 如果 True 为真，则打印诊断消息。
  2. REPUB_COUNT 用于调试目的。记录自启动以来使用相同 PID 发生的重新发布总次数。

3.4 模块属性

  1. VERSION 一个包含三个 int 的 3 元组（major, minor, micro），例如 (0, 5, 0)。

3.5 基于事件的接口

通过设置 config["queue_len"] = N 来调用此接口，其中 N > 0。在此 模式下没有回调。传入的消息会被排队，并可通过异步迭代器访问。 该模块通过设置绑定的 .up 和 .down Event 实例来报告连接状态变化。 演示程序 range.py 和 range_ex.py 使用了此接口。以下代码片段展示了其用法。

读取消息：

async def messages(client):
    # If MQTT V5 is in use
    # async for topic, msg, retained, properties in client.queue:
    async for topic, msg, retained in client.queue:
        print(f'Topic: "{topic.decode()}" Message: "{msg.decode()}" Retained: {retained}')

处理连接事件：

async def up(client):  # (re)connection.
    while True:
        await client.up.wait()
        client.up.clear()
        print('We are connected to broker.')
        await client.subscribe('foo_topic', 1)  # Re-subscribe after outage

可选的中断处理程序：

async def down(client):
    while True:
        await client.down.wait()  # Pause until outage
        client.down.clear()
        print('WiFi or broker is down.')

使用默认小队列进行初始化：

config["queue_len"] = 1
client = MQTTClient(config)

在传入消息到达缓慢且 clean 标志为 False 的应用中，消息不会累积，队列长度可以很小。 值为 1 可提供最小队列。当消息突发到达且速度过快， 导致应用无法及时处理时，队列就会发挥作用。这种情况可能 发生在多个客户端独立发布到同一主题时：代理服务器可能 以高速度将它们转发给订阅者。另一种情况是当 clean 标志为 False 且发生长时间 wifi 中断时：当中断结束时，可能 会有大量积压的消息。此类情况可能需要更大的队列。

如果队列溢出，最旧的消息将被丢弃。 此策略优先考虑弹性而非 qos==1 保证。绑定的 变量 client.queue.discards 会保持丢失消息的累计总数。在 开发过程中，这有助于确定最佳队列长度。

有可能（尽管很少有用）让多个任务等待消息。这些任务必须在每条消息之后让出控制权，以便其他任务能够被调度。消息将以轮询方式在等待的任务之间分发。可以创建此任务的多个实例：

async def messages(client):
    async for topic, msg, retained in client.queue:
        await asyncio.sleep(0)  # Allow other instances to be scheduled
        # handle message

在应用程序中，RAM 非常宝贵，在测试中，基于回调的接口比最小队列情况消耗略低（约 1.3KiB）。

目录

3.6 MQTTv5 支持

应用程序设计人员应考虑 V5 是否合适。相关因素包括：

  • 如果客户端要订阅由 V5 客户端发布的消息，且消息中包含必须接收的属性，则 V5 是必需的。
  • 如果客户端要以干净会话 False 运行，V5 提供重大优势。 这里的意图是确保在连接不可靠的情况下，消息不会在短暂中断期间丢失。在 V3.1.1 中，这有一个不幸的后果，即当发生长时间中断时，可能会有大量积压。 V5 消息可以带有“消息过期间隔”属性发布，提供了一种限制积压大小的选项。这在连接时设置的“会话过期间隔”下得到进一步增强。
  • 如果有发布带有属性的消息的需求，则 V5 是必需的。
  • 如果 RAM 有限，V5 会使库的使用量增加约 3KiB。
  • V5 启用请求/响应协议，其中传入的消息可以与特定的发布相关联。

3.6.1 配置及从 MQTTv3.1.1 迁移

支持 MQTTv5，并可通过在 config 字典中将 mqttv5 设置为 True 进行配置。默认值为 False。V5 应用程序不应使用 clean_init 配置值 - 消息积压应使用 会话和消息过期间隔来控制。支持连接时的属性，这些 属性需要在配置字典中传入。有关属性的更多信息 及其格式，请参阅 3.6.2。

from mqtt_as import MQTTClient, config
config['mqttv5'] = True

# Optional: Set the properties for the connection
config['mqttv5_con_props'] = {
    0x11: 3600,  # Session Expiry Interval
}

# The rest of the configuration
client = MQTTClient(config)

API 进行了修改以支持 MQTTv5 功能。最 重要的是添加了 properties 参数，该参数作为 事件和基于回调的消息处理器的额外参数提供。

# For MQTT 3.1.1 support
def callback(topic, msg, retained):
    print((topic, msg, retained))

# For MQTT 5 and 3.1.1 support
def callback(topic, msg, retained, properties=None):
    print((topic, msg, retained, properties))

允许将属性作为可选参数，使您能够在不更改回调签名的情况下，在 MQTT 3.1.1 和 MQTT 5 支持之间进行切换。

async def messages(client):
    async for topic, msg, retained, properties in client.queue:
        await asyncio.sleep(0)  # Allow other instances to be scheduled
        # handle message

properties 参数是一个包含消息属性的字典。这些属性在 MQTTv5 规范中定义。如果在使用 MQTTv3.1.1 时在发布消息中包含属性，这些属性将被忽略。

3.6.2 MQTTv5 属性

属性是 MQTTv5 的一项新且重要的特性。它们用于提供关于消息的附加信息，并支持更高级的功能，例如消息过期、用户属性和响应信息。

传入的属性格式化为一个字典，使用属性标识符作为键。属性标识符是一个在 MQTTv5 规范中定义的整数，模块中未为这些值定义常量。属性标识符定义在 MQTTv5 规范。

发送属性必须采用正确的格式。MQTTv5 规范区分二进制属性和文本属性。确保属性以正确格式发送非常重要。作为参考，请参阅 MQTTv5 规范的 第 2.2.2.2 节。

properties = {
    0x26: {'value': 'test'},       # User Property (UTF-8 string pair)
    0x09: b'correlation_data',     # Correlation Data (binary)
    0x08: 'response_topic',        # Response Topic (UTF-8 string)
    0x02: 60,                      # Message Expiry Interval (integer)
}

await client.publish('topic/test', 'message', False, 0, properties=properties)

在下表中，属性类型被定义为 Python 变量类型；"string" 是 utf8 编码的 str。V5 协议提供了一种可选的请求/响应交换机制。这在 V5 规范的 第 4.10 节 中有描述。

传出属性

以下是与 client.publish() 相关的属性摘要。

KeyValueDestinationNameMeaning
0x01bytesubscriberpayload format indicator0=binary 1=utf8
0x02intbrokerMessage Expiry IntervalLifetime in seconds
0x03stringsubscriberContent TypeApplication defined
0x08stringsubscriberResponse TopicRequest/response
0x09bytessubscriberCorrelation DataRequest/response
0x23intbrokertopic alias
0x26string pairsubscriberuser propertyApplication defined

与 client.subscribe() 相关的属性：

KeyValueNameMeaning
0x0BintSubscription IdentifierSee below
0x26string pairuser propertyApplication defined

订阅标识符使客户端应用程序能够将其当前状态传递给 broker：对该订阅的响应将包含该状态。参见 规范 section 3.8.4

与 client.connect() 相关的属性。请注意，连接属性在配置字典中提供（config[mqttv5_con_props]）。所有属性均为对 broker 的指令。除 0x11 和 0x22 外，其余属性均较为晦涩，需要研究规范。

KeyValueNameMeaning
0x11intSession expiry interval (secs)See below
0x17byte 0/1Request Problem InformationSpec 3.1.2.11.7
0x19byte 0/1Request Response InformationSpec 3.1.2.11.6
0x21intReceive MaximumSpec section 4.9 etc
0x22intTopic Alias MaximumMax alias value
0x26string pairuser propertyApplication defined
0x27intMaximum Packet SizeSpec 3.2.2.3.6

Session Expiry Interval 定义了 broker 在断开连接后保留会话状态的时间。将其设置为零并设置 clean=True 等同于 MQTT3.1.1 中的 Clean Session 状态：在故障期间发生的发布将被错过。较长的过期间隔允许接收此类消息，但在长时间故障后存在大量积压的风险。

传入属性
键值来源名称含义
0x01bytepublisherpayload format indicator0=二进制 1=utf8
0x02intpublisherMessage Expiry Interval以秒为单位的生存时间
0x03stringpublisherContent Type应用程序定义
0x08stringpublisherResponse Topic请求/响应
0x09bytespublisherCorrelation Data请求/响应
0x0BintpublisherSubscription Identifier
0x26string pairpublisheruser property应用程序定义

从 broker 接收到的其他数据包可能包含属性。除了 CONNACK 之外，mosquitto broker 似乎仅在错误 条件下发送这些属性。无论如何，如果 MQTTClient.DEBUG 为 True，客户端会打印这些属性。示例数据包为 CONNACK、PUBACK、SUBACK、UNSUBACK 和 DISCONNECT。

Topic Alias

主题别名的核心思想是减少出站消息的大小。 发布时使用完整的主题名称，并将主题别名设置为一个 非零整数。后续的发布可以传递主题为 ""，并将主题别名属性设置为该整数。 关键是代理必须接收到设置别名的消息，因为如果收到未知别名，它可能会断开连接。 如果在发布别名消息期间发生故障，也会出现一个问题。 似乎代理在故障后不会存储别名。 在重新连接时重新建立任何别名是应用程序的责任。 请在 规范 使用此功能之前进行研究。

3.6.3 不支持的功能

为了保持库的轻量级和经过充分测试，MQTTv5 的某些 功能不受支持。

  1. 增强身份验证：增强身份验证 是 MQTT 规范中新增的一部分，允许使用更高级的 身份验证方法。该库不支持此功能。 AUTH 数据包未实现且未被处理。
  2. 遗嘱属性：遗嘱属性 随着 MQTTv5 消息中属性的引入，现在可以拥有属性。 这包括遗嘱消息。此功能不受支持，因此 无法随遗嘱消息发送属性。
  3. 多个用户属性：用户属性 规范允许随消息发送多个用户属性。在 当前实现中，仅支持一个用户属性。这适用于 发送和接收消息。在接收消息时，仅返回最后一个用户 属性。如果在发送消息时在用户 属性字典中包含超过 1 个键值对，则仅 发送第一个键值对。
  4. 订阅选项：订阅选项 在 MQTTv5 中引入了订阅选项（除 QoS 级别外）。 在订阅主题时无法设置这些选项。以下选项 不可用：
    • No Local (NL)
    • Retain As Published (RAP)
    • Retain Handling
  5. CONNACK 数据包中的并非所有属性都已公开。
  6. Properties on operations other than CONNECT and PUBLISH are not returned to the user. For more information, see this comment
  7. The client does not store incoming Topic Alias properties.

注意：这些功能中的大多数都可以通过一些努力来实现。 这些功能未被实现，以保持当前实现的简洁性 并减少所需的测试范围。

目录

4. 备注

4.1 连接性

如果在构造函数调用中定义了 keepalive，则代理将假设 如果在该时间段内未收到任何消息，则连接已丢失。 该模块通过最多在 keepalive 间隔期间发送四次 MQTT ping 来尝试保持连接打开。 （如果来自代理的最后一次响应超过 keepalive 周期的 1/4，则发送 ping）。 更频繁的 ping 可能有助于降低故障检测的延迟。 这可以使用 ping_interval 配置选项来完成。 这里的要点是，虽然 WiFi 故障可以快速检测到，但上游故障只能通过 来自代理的通信缺失来检测。如果 ping 间隔较长，代理可能 在客户端检测到并发起重连尝试之前长时间无法访问。

如果代理超时，它将发布“最后遗嘱”消息（如果有的话）。 这将由订阅该主题的其他客户端接收。

如果客户端确定连接已丢失，它将关闭 套接字并定期尝试重新连接，直到成功为止。

在连接失败的情况下，客户端和服务器发布的 qos == 0 消息可能会丢失。qos == 1 数据包的行为如下所述。

4.2 客户端 qos 1 发布

这些行为如下。客户端等待 response_time。如果未收到确认，它将重新发布，最多 MAX_REPUBS 次。 在没有确认的情况下，假定网络已断开。客户端将按照上述描述重新连接。然后，发布将作为一条具有不同 PID 的新消息再次尝试。（事实证明，新的 PID 对于 Mosquitto 识别该消息是必要的）。

这实际上保证了 qos == 1 发布的接收，前提是发布协程将阻塞，直到接收被确认。

允许 qos == 1 的发布在等待确认暂停的同时并发运行，然而这对资源受限的设备有影响。参见 第 4.4 节。

4.3 客户端 qos 1 订阅

当客户端以 qos == 1 订阅某个主题，并且发生 qos == 1 的发布时，代理将重新发布，直到收到确认。如果代理认为连接已失败，它将等待客户端重新连接。如果客户端配置为将 clean 设置为 True， 则在故障期间发布的 qos == 1 消息将丢失。否则，它们将快速连续接收（这可能会溢出 ESP8266 上的缓冲区，导致 LmacRxBlk:1 消息）。

4.4 应用设计

该模块允许并发发布和订阅注册。

当在资源有限的硬件（如 ESP8266）上使用 qos == 1 发布时，明智的做法是通过实现单个发布任务来避免并发。在这种情况下，如果需要发布队列，应由应用程序来实现。

在性能较强的硬件上，让多个协程异步执行 qos == 1 发布是有效的，但与 broker 的连接缓慢时会有影响：等待 PUBACK 数据包的任务堆积意味着资源的消耗。

WiFi 和 Connect 协程相对于连接和断开网络所需的时间应快速运行至完成。目标为最多 2 秒。或者，Connect 协程可以无限期运行，只要当 isconnected() 方法返回 False 时它会终止。

订阅回调会阻塞发布以及后续订阅消息的接收，因此应设计为快速返回。

4.4.1 发布超时

一位贡献者（Kevin Köck）担心，在连接中断的情况下，发布可能会被延迟到过度过时的程度。他希望实现一个超时机制，如果中断导致高延迟，则取消发布。这可以说是 MQTT3.1.1 的一个限制 - 请参阅 MQTTv5 支持。

以下说明是对 V3.1.1 变通方案的讨论。

简单取消发布任务是不推荐的，因为这可能会破坏 MQTT 协议。有几种方法可以解决此问题：

  1. 在发布时发送时间戳，订阅者在收到延迟消息时采取相应措施。
  2. 在发布前检查连接状态。这并非绝对可靠，因为连接可能在检查之后、发布开始之前失败。
  3. 子类化 MQTTClient 并在发出取消请求之前获取 self.lock 对象。self.lock 对象保护协议序列，使其不会受到其他任务的干扰。这是已成功采用的方法，可以在 mqtt_as_timeout.py 中看到。

这未包含在库中，主要是因为大多数用例都可以通过使用时间戳来覆盖。其他原因已在代码注释中记录。

4.4.2 上电时的行为

该库旨在透明地处理连接中断，但客户端的电源循环必须在应用层面加以考虑。当应用程序调用客户端的 connect 方法时，任何失败都会引发 OSError。这是有意为之，因为所采取的操作取决于应用程序。可能需要检查 WiFi 或代理功能。可能需要回退到不同的网络。在其他应用程序中，可能会预期短暂的电源中断：当电源恢复时，客户端将简单地重新连接。如果发生错误，应用程序可能会等待一段时间后再重试。

当初始连接建立后，由网络中断引起的后续连接对应用程序是透明处理的。

在此上下文中，应考虑 "clean session"（干净会话）的行为。如果 clean 标志为 False 且发生长时间断电，可能会产生大量 积压消息。这会在资源受限的客户端上引发问题， 尤其是当客户端已停用数天时。MQTTv5 对此进行了优雅的处理 - 请参阅 MQTTv5 支持。

对于使用 MQTTv3 的用户，本模块通过启用在 上电情况与网络中断情况下行为不同的机制来解决此问题。

clean_init 标志决定上电时的行为，而 clean 定义 连接中断后的行为。如果 clean_init 为 True 且 clean 为 False，上电时将丢弃先前的会话状态。客户端将以 clean==False 重新连接。在连接中断后，它也会以类似方式重新连接。因此， 上电后，订阅将满足在连接中断期间发布的消息的 qos==1 保证。

如果两个标志均为 False，则会出现正常的非干净会话行为， 长时间断电后可能会产生大量积压消息。

如果上电时两个标志均为 True，代理将在连接（因此也是断电） 中断期间丢弃会话状态。这意味着在连接中断期间发布的消息 将会丢失（MQTT 规范 3.1.2.4 干净会话）。

此处也有讨论 这里。

4.5 替代设计方法

以下方法将 MQTT 发布-订阅模型扩展到 asyncio 应用程序中。这提供了一种替代的应用程序设计方式，其中 消息传递是主要的控制机制。 消息代理 会被实例化。传入的 MQTT 消息会被转发到消息代理。 任务可以订阅 Broker 实例，使得消息触发一个 动作。参见 async_message.py 获取示例。要运行代码，需要 安装 asyncio 原语：

$ mpremote mip install "github:peterhinch/micropython-async/v3/primitives"

在此演示中，MQTT 消息被发布到主题 "red_topic" 和 "blue_topic"。 消息内容为 "on" 或 "off"。接收 messages 任务将所有传入 消息转发到 Broker 实例。应用程序可以 以多种方式订阅主题。在此演示中，它订阅了一个函数 led_handler 到 这两个主题；该函数根据消息文本控制传入的 LED。

以下说明了 async_message.py 如何实现这一点：

from mqtt_as import MQTTClient
from mqtt_local import wifi_led, blue_led, config
import asyncio
from primitives import Broker

# Incoming "red_topic" and "blue_topic" messages are directed to led_handler
def led_handler(topic, message, led):
    led(message == "on")

broker = Broker()
# Subscribe led_handler function to the two topics
broker.subscribe("blue_topic", led_handler, blue_led)
broker.subscribe("red_topic", led_handler, wifi_led)

# All incoming MQTT messages are forwarded to the Broker
async def messages(client):
    async for topic, msg, retained in client.queue:
        broker.publish(topic.decode(), msg.decode())

config["queue_len"] = 1  # Must use event interface

可以订阅除函数以外的对象，包括协程、 方法、队列、Event 实例和用户定义的类实例。

目录

4.4.3 优化

版本 0.8.2 引入了一项优化，将传入消息读取到 预分配的缓冲区中。这避免了内存分配并提高了性能。 该更改以不破坏现有代码的方式完成。通过设置两个模块变量， 可以进一步减少内存分配。这些变量为（默认值）：

IBUFSIZE = 50 MSG_BYTES = True

所有更改都应在实例化客户端之前进行，例如：

import mqtt_as

mqtt_as.IBUFSIZE = 5_000
client = MQTTClient(config)
IBUFSIZE

套接字读取操作会写入一个预分配的缓冲区。如果到达的消息过大，缓冲区将被扩展以容纳它。这意味着会发生内存分配。考虑这样一种情况：在长时间仅接收短消息之后，一条长消息到达。此时 RAM 可能已经变得碎片化，导致大型内存分配失败。如果已知可能会到达大型消息，在初始阶段（即碎片化发生之前）设置较大的缓冲区大小可以避免此问题。

MSG_BYTES

默认情况下，传入的消息在提供给应用程序之前会被复制。这意味着会发生内存分配。这样做是为了确保在所有条件下消息的完整性。如果使用事件接口，无论 MSG_BYTES 如何，都会发生复制。

在使用回调接口且 MSG_BYTES 为 False 的情况下，会将缓冲区的 memoryview 传递给回调，从而避免内存分配。 如果返回 memoryview，以下代码是安全的：

# Subscription callback
def sub_cb(topic, msg, retained):
    # Synchronous code handles the message

然而，如果返回 memoryview，则存在隐患：

# Subscription callback
def sub_cb(topic, msg, retained):
    asyncio.create_task(process_message(topic, msg))

如果在 process_message 完成之前另一条消息到达，就会发生故障。 缓冲区内容将会改变，导致数据损坏。

Contents

5. 非标准应用程序

mqtt_as 的正常运行基于尽可能保持链路连接。这确保了订阅的最小延迟，但意味着会消耗电力。machine 模块支持两种省电模式：lightsleep 和 deepsleep。目前 asyncio 不支持这两种模式。以下说明可能适用于任何有意关闭并重新打开与 broker 链路的应用程序。

5.1 deepsleep

通过定期连接、处理发布和待处理的订阅，然后进入 deepsleep，可以实现最大的省电效果。借助合适的硬件，可以制作出平均功耗极低的 MQTT 客户端。这是通过缩短应用程序运行时间并使用 machine.deepsleep 进行一段时间睡眠来实现的。当时间段到期时，板子 重置，main.py 重新启动应用程序。

测试的硬件是 UM Feather S2，可从 Adafruit 获得。我的样品在 deepsleep 模式下仅消耗 66μA 电流。它具有可切换的 LDO 稳压器，允许在主机处于 deepsleep 时关闭外部传感器的电源。它还支持通过 LiPo 电池组进行电池 操作，并可通过 USB 充电。配备 WBUS-DIP28 的 Pyboard D 具有 类似的特性。

测试脚本 lptest_min.py 会定期唤醒并连接到 WiFi。它发布来自 板载光传感器的值，并订阅主题 "foo_topic"。在 deepsleep 期间发生的任何匹配 发布都会被接收，并通过闪烁蓝色 LED 显示出来。

请注意，deepsleep 会禁用 USB。这在开发过程中会带来不便。 该脚本具有测试模式，在此模式下，deepsleep 被替换为 time.sleep 和 machine.soft_reset，以保持 USB 链路处于活动状态。调试的另一种方法是 使用带有 FTDI 适配器的 UART。此类链路可以在 深度睡眠期间保持存活。

每次客户端进入 deepsleep 时，都会发出 .disconnect()。这会向 broker 发送一个 MQTT DISCONNECT 数据包，根据 MQTT 规范第 3.1.2.5 节抑制 last will。其理由是，deepsleep 的持续时间很可能 远长于 keepalive 时间。使用 .disconnect() 可确保仅在发生诸如程序 崩溃等故障时才触发 last will 消息。

在关闭连接并进入 deepsleep 的应用程序中，通过将 quick 参数设置为 .connect， 可以进一步降低功耗。在断线后连接或重新连接时，会进行检查以确保 WiFi 连接的稳定性。快速连接仅在初始连接时跳过此检查，从而 节省几秒钟。这里的理由是，初始连接中的任何错误 都必须由应用程序处理。测试脚本在重新尝试连接前会休眠 retry 秒。

5.2 lightsleep 和断开连接

该库并非设计用于系统进入 lightsleep 的情况。首先，asyncio 并不支持所有平台上的 lightsleep - 特别是在 STM 上，ticks_ms 时钟（对任务调度至关重要）会在 lightsleep 期间停止。

其次，该库没有机制来确保在发出 .disconnect 后所有任务都能干净地关闭。 这让人对任何发出 .disconnect 然后尝试重新连接的应用程序产生质疑。 使用 deepsleep 时不会出现此问题，因为主机实际上会断电。 当睡眠结束时，asyncio 和必要的任务会像上电事件一样启动。

这些问题已被用户通过该库的分支为特定应用程序解决。 鉴于 asyncio 的局限性，我不打算编写通用解决方案。

5.3 超低功耗

本文档 描述了一种用于 ESP32 或 ESP8266 的 MQTT 客户端，它使用 ESPNOw 与运行 mqtt_as 的网关进行通信。 客户端每次唤醒时无需重新连接 WiFi，从而节省电量。 该网关可在多个客户端之间共享。

缺点是需要一个始终在线的网关，并且仅支持 MQTT V3.1.1 功能的一个子集。

目录

6. 参考文献

mqtt 介绍
mosquitto 服务器
mosquitto 客户端发布
mosquitto 客户端订阅
MQTT 3.1.1 规范
MQTTv5 规范
用于 PC 的 python 客户端
非官方 MQTT 常见问题解答
公共代理列表

目录

7. 连接错误代码

在初始连接尝试中，broker 可能会拒绝该尝试。在此 情况下，将引发一个 OSError，其中显示两个数字。第一个数字 应为 0x2002，即 MQTT CONNACK 固定头部。第二个 是 CONNACK 可变头部字节 2，其指示失败原因如下：

值原因
1不可接受的协议版本。
2客户端标识符被拒绝。
3MQTT 服务不可用。
4用户名或密码格式无效。
5客户端未授权连接。

参见 MQTT 规范第 3.2.2 节。

目录

8. Hive MQ

Hive MQ 网站提供一个免费的基于 Web 的 broker， 其安全性高于公共 broker。使用公共 broker 时，任何人都可以 检测并订阅您的发布内容。Hive MQ 为您提供一个唯一的 broker 互联网地址，访问该地址需要密码。TLS 是强制性的，但不 需要证书。

简单的 GitHub 注册即可获得：

  • 唯一的 broker 地址。
  • 您指定一个用户名。
  • 网站提供一个密码。

典型用法：

config['user'] = 'my_username'
config['password'] = 'my_password'
broker = 'unique broker address'  # e.g long_hex_string.s2.eu.hivemq.cloud
config['server'] = broker
config['ssl'] = True
config['ssl_params'] = {"server_hostname": broker}

该免费服务可扩展（需付费）至大型商业部署。

目录

9. ssl_params 字典

以下是允许的键：

根据 此帖子 以下平台使用 mbedtls：

  • esp32 端口
  • pico w
  • unix 端口
  • stm32

参见 此帖子 了解如何使用客户端证书与 mosquitto 代理服务器配合使用。

请注意，使用客户端证书的 TLS 要求客户端时钟大致准确。 这可以通过 NTP 查询实现。如果在本地服务器上运行 mosquitto， 它也会运行 NTP 守护进程。一种高可用性选项是对本地服务器执行 NTP 查询。参见 此文档， 以及 官方 ntptime 模块。

参见 此链接 以获取 有关创建客户端证书及用于此操作的 Bash 脚本的信息。

参见 此网站， 其中包含大量关于 SSL 的有用信息。