asyncio_mqtt订阅多个MQTT主题仅第一个生效的问题求助

你实现时遇到只有第一个订阅主题生效的核心原因是：filtered_messages异步上下文的async for迭代是阻塞式的，如果你把订阅逻辑写在for循环内部，第一个主题的消息迭代启动后，后续的订阅逻辑根本没有执行机会，因此只有第一个主题能正常接收消息。

以下是两种可行的实现方案：

方案1：直接使用全局消息队列（最简单） #

不需要单独做消息过滤，只要先完成所有主题的订阅，直接消费客户端全局的消息队列即可，所有订阅主题的消息都会进入这个队列：

import asyncio
from asyncio_mqtt import Client

class mqtt_task():
    def __init__(self, broker, subscription):
        self.broker = broker
        self.subscription = subscription # 存储主题的列表

    async def sub_and_listen_task(self):
        while True:
            try:
                async with Client(self.broker) as client:
                    self.client = client
                    # 直接传入主题列表完成批量订阅，可传入第二个参数对应QoS等级列表
                    await client.subscribe(self.subscription)
                    # 消费所有已订阅主题的消息
                    async for message in client.messages:
                        print(message.payload.decode(), ' ' , message.topic)
            except Exception as e:
                print(f"连接异常，5秒后重连：{e}")
                await asyncio.sleep(5)

# 启动示例
if __name__ == "__main__":
    broker = "替换为你的MQTT broker地址"
    subscription = ["topic1", "topic2/test", "topic3/#"]
    task = mqtt_task(broker, subscription)
    asyncio.run(task.sub_and_listen_task())

方案2：按主题分逻辑处理（不同主题需要走不同处理逻辑时使用） #

如果不同主题的消息需要走独立的处理逻辑，就给每个主题的过滤队列开单独的协程任务，用asyncio.gather同时运行所有任务：

import asyncio
from asyncio_mqtt import Client

class mqtt_task():
    def __init__(self, broker, subscription):
        self.broker = broker
        self.subscription = subscription

    # 单个主题的消息处理协程
    async def handle_topic(self, client, topic):
        async with client.filtered_messages(topic) as messages:
            async for message in messages:
                # 此处编写对应主题的专属处理逻辑
                print(f"主题[{topic}]收到消息：{message.payload.decode()}")

    async def sub_and_listen_task(self):
        while True:
            try:
                async with Client(self.broker) as client:
                    self.client = client
                    # 先完成所有主题的批量订阅
                    await client.subscribe(self.subscription)
                    # 为每个主题创建独立的处理任务
                    tasks = [asyncio.create_task(self.handle_topic(client, topic)) for topic in self.subscription]
                    # 同时运行所有处理任务
                    await asyncio.gather(*tasks)
            except Exception as e:
                print(f"连接异常，5秒后重连：{e}")
                await asyncio.sleep(5)

if __name__ == "__main__":
    broker = "替换为你的MQTT broker地址"
    subscription = ["topic1", "topic2/test", "topic3/#"]
    task = mqtt_task(broker, subscription)
    asyncio.run(task.sub_and_listen_task())

注意事项 #

• 如果你使用的是paho-mqtt的异步接口而非asyncio-mqtt，批量订阅的传参格式为列表嵌套元组，比如[("topic1", 0), ("topic2", 1)]，分别对应主题和QoS等级
• 订阅时如果使用通配符#、+要符合MQTT主题规范，避免匹配到不需要的消息
• 重连逻辑中要注意每次重连都要重新执行所有主题的订阅操作

内容的提问来源于stack exchange，提问作者ajay1738