asyncio-mqtt - 异步 MQTT 客户端

• 1. 项目简介
• 2. 快速开始
• 3. 客户端配置
• 4. 最佳实战
  • 4.1 重新连接
  • 4.2 取消任务
  • 4.3 分享客户端
  • 4.4 过滤请求
  • 4.5 不阻塞运行
  • 4.6 TLS
  • 4.7 代理

1. 项目简介

项目	信息
项目地址	PyPI | GitHub
官方文档	sbtinstruments.github.io
开源协议	BSD-3-Clause
Python 版本	Python 3.7 ~ 3.11
标签	asyncio

asyncio-mqtt 提供了基于 paho-mqtt 的现代的、异步的接口。

• 没有回调
• 没有更多的返回代码（使用 MqttError 就够了）
• 优雅的断开连接
• 与异步代码兼容
• 完全的类型提示
• 整个系统只有不到 900 行的代码。

asyncio-mqtt 安装后同样也会安装 paho.mqtt，部分配置可能用到 paho.mqtt 库中的内容。

安装：

pip install asyncio-mqtt

直接安装 GitHub 上的最新版本：

pip install git+https://github.com/sbtinstruments/asyncio-mqtt

Windows 使用注意

从 Python 3.8 开始，asyncio 的默认事件循环是 ProactorEventLoop。该循环不支持 asyncio-mqtt 需要的 add_reader 方法。请切换到一个支持 add_reader 方法的事件循环，比如内置的 SelectorEventLoop。

# 修改为 Windows 的 "Selector" 事件循环
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# 正常使用即可
asyncio.run(main())

2. 快速开始

发布：

async with Client("test.mosquitto.org") as client:
    await client.publish("humidity/outside", payload=0.38)

订阅：

async with Client("test.mosquitto.org") as client:
    async with client.messages() as messages:
        await client.subscribe("humidity/#")
        async for message in messages:
            print(message.payload)

payload 可以是 int、float、str、bytes、bytearray 和 None。

数字被默认转换为字符串，如果需要数字原始值，可以使用 struct.pack()。None 代表零长度负载。

3. 客户端配置

在初始化客户端时，可参考下面的配置，下面的值均为默认值。关于各个参数的更多信息，请看 paho-mqtt 的文档。

import asyncio_mqtt as aiomqtt
import paho.mqtt as mqtt

aiomqtt.Client(
    hostname="test.mosquitto.org",  # 唯一的必须参数
    port=1883,
    username=None,
    password=None,
    logger=None,
    client_id=None,
    tls_context=None,
    tls_params=None,
    proxy=None,
    protocol=None,
    will=None,
    clean_session=None,
    transport="tcp",
    keepalive=60,
    bind_address="",
    bind_port=0,
    clean_start=mqtt.client.MQTT_CLEAN_START_FIRST_ONLY,
    properties=None,
    message_retry_set=20,
    socket_options=(),
    max_concurrent_outgoing_calls=None,
    websocket_path=None,
    websocket_headers=None,
)

4. 最佳实战

4.1 重新连接

import asyncio
import asyncio_mqtt as aiomqtt

async def main():
    reconnect_interval = 5  # In seconds
    while True:
        try:
            async with aiomqtt.Client("test.mosquitto.org") as client:
                async with client.messages() as messages:
                    await client.subscribe("humidity/#")
                    async for message in messages:
                        print(message.payload.decode())
        except aiomqtt.MqttError as error:
            print(f'Error "{error}". Reconnecting in {reconnect_interval} seconds.')
            await asyncio.sleep(reconnect_interval)

asyncio.run(main())

4.2 取消任务

@3.11+ Python 3.11 新增了 asyncio.timeout，可用监听一段时间的消息。

import asyncio
import asyncio_mqtt as aiomqtt

async def listen():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("humidity/#")
            async for message in messages:
                print(message.payload)

async def main():
    try:
        # 5 秒后取消请求
        async with asyncio.timeout(5):
            await listen()
    # 忽略 TimeoutError
    except asyncio.TimeoutError:
        pass

asyncio.run(main())

Python 3.11 之前的版本可用 task.cancel 来取消：

import asyncio
import asyncio_mqtt as aiomqtt

async def listen():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("humidity/#")
            async for message in messages:
                print(message.payload)

async def main():
    loop = asyncio.get_event_loop()
    # 创建任务
    task = loop.create_task(listen())
    # 等待特定的任务
    await asyncio.sleep(5)
    # 取消任务
    task.cancel()
    # 等待任务取消
    try:
        await task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

4.3 分享客户端

import asyncio
import asyncio_mqtt as aiomqtt

async def publish_humidity(client):
    await client.publish("humidity/outside", payload=0.38)

async def publish_temperature(client):
    await client.publish("temperature/outside", payload=28.3)

async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await publish_humidity(client)
        await publish_temperature(client)

asyncio.run(main())

4.4 过滤请求

import asyncio
import asyncio_mqtt as aiomqtt

async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("#")
            async for message in messages:
                if message.topic.matches("humidity/outside"):
                    print(f"[humidity/outside] {message.payload}")
                if message.topic.matches("+/inside"):
                    print(f"[+/inside] {message.payload}")
                if message.topic.matches("temperature/#"):
                    print(f"[temperature/#] {message.payload}")

asyncio.run(main())

4.5 不阻塞运行

使用 asyncio.TaskGroup（或 asyncio.gather，要求 Python < 3.11）来运行多个任务：

import asyncio
import asyncio_mqtt as aiomqtt

async def sleep(seconds):
    await asyncio.sleep(seconds)
    print(f"Slept for {seconds} seconds!")

async def listen():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("humidity/#")
            async for message in messages:
                print(message.payload)

async def main():
    async with asyncio.TaskGroup() as group:
        group.create_task(sleep(2))
        group.create_task(listen())  # 此处开始监听
        group.create_task(sleep(3))
        group.create_task(sleep(1))

asyncio.run(main())

如果需要和其他异步 Web 框架等结合，我们有下面的方案：

import asyncio
import asyncio_mqtt as aiomqtt

async def listen():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        async with client.messages() as messages:
            await client.subscribe("humidity/#")
            async for message in messages:
                print(message.payload)

background_tasks = set()

async def main():
    loop = asyncio.get_event_loop()
    # 创建任务
    task = loop.create_task(listen())
    # 创建引用以避免被垃圾回收
    background_tasks.add(task)
    task.add_done_callback(background_tasks.remove)

    # 持续运行别的任务
    while True:
        await asyncio.sleep(2)

asyncio.run(main())

4.6 TLS

import asyncio
import asyncio_mqtt as aiomqtt
import ssl

tls_params = aiomqtt.TLSParameters(
    ca_certs=None,
    certfile=None,
    keyfile=None,
    cert_reqs=ssl.CERT_REQUIRED,
    tls_version=ssl.PROTOCOL_TLS,
    ciphers=None,
)

async def main():
    async with aiomqtt.Client("test.mosquitto.org", tls_params=tls_params) as client:
        await client.publish("humidity/outside", payload=0.38)

asyncio.run(main())

4.7 代理

import asyncio
import asyncio_mqtt as aiomqtt
import socks

proxy_params = aiomqtt.ProxySettings(
    proxy_type=socks.HTTP,
    proxy_addr="www.example.com",
    proxy_rdns=True,
    proxy_username=None,
    proxy_password=None,
)

async def main():
    async with aiomqtt.Client("test.mosquitto.org", proxy=proxy_params) as client:
        await client.publish("humidity/outside", payload=0.38)

asyncio.run(main())