如何用aio-pika从多队列获取消息并动态声明队列？

问题1：当前手动声明多队列的方法是否正确？ #

你的手动实现功能上是正确的：

• 独立调用channel.declare_queue()声明每个持久化队列，符合RabbitMQ的队列声明规范
• 对每个队列调用queue.consume(on_message)绑定消息处理函数，能正常监听多个队列的消息
• 最后用await asyncio.Future()维持事件循环运行，避免程序直接退出

不过代码存在重复冗余，仅适合队列数量固定的场景；如果队列数量需要动态调整，就需要优化成批量处理的方式。

问题2：如何根据队列名列表动态声明队列？ #

可以通过异步遍历队列名列表实现动态声明，步骤如下：

1. 实现从数据库获取队列名的异步函数（替换为你的实际数据库查询逻辑）
2. 遍历队列名列表，批量完成队列声明与消息监听绑定
3. 保持事件循环持续运行

示例代码如下：

import asyncio
from aio_pika import connect    

# 模拟从数据库获取队列名列表的异步函数
async def get_queue_names_from_db() -> list[str]:
    # 实际场景替换为你的数据库异步查询逻辑
    return ["queue_1", "queue_2", "queue_3", "queue_4"]

async def on_message(message):
    # 你的消息处理逻辑
    async with message.process():
        print(f"Received message from {message.queue}: {message.body.decode()}")

async def main() -> None:
    connection = await connect(Settings.RABBIT_URL)
    
    async with connection:
        channel = await connection.channel()
        await channel.set_qos(prefetch_count=0)
        
        # 获取动态队列名列表
        queue_names = await get_queue_names_from_db()
        
        # 遍历列表，动态声明队列并绑定消费
        for queue_name in queue_names:
            queue = await channel.declare_queue(
                queue_name,
                durable=True,
            )
            await queue.consume(on_message)
        
        print(" [*] Waiting for messages. To exit press CTRL+C")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

如果偏好批量执行的写法，也可以用asyncio.gather一次性完成所有队列声明：

# 批量声明队列
queues = await asyncio.gather(*[
    channel.declare_queue(name, durable=True)
    for name in queue_names
])

# 批量绑定消费逻辑
for queue in queues:
    await queue.consume(on_message)

注意：从数据库获取队列名的操作必须是异步的，避免阻塞事件循环；如果队列名可能在程序运行中动态变化，还需要额外实现监听数据库变更、动态新增/移除队列的逻辑（比如定时轮询数据库或使用数据库触发器）。

内容的提问来源于stack exchange，提问作者Ibnu Agiels Althur