如何使用aio-pika创建全局RabbitMQ连接（非函数内）并连接到队列？

当然可以实现，下面是两种常用的方案，帮你在函数外部创建全局RabbitMQ连接并关联队列：

方案一：模块级全局变量 + 异步初始化 #

通过定义模块级别的全局变量存储连接、通道和队列，再通过异步初始化函数在程序启动时完成创建，之后就可以在任意函数中直接使用这些全局变量。

示例代码：

import aio_pika
from aio_pika import Connection, Channel, Queue

# 全局变量，用于存储连接、通道和队列
global_connection: Connection | None = None
global_channel: Channel | None = None
global_queue: Queue | None = None

async def init_rabbitmq():
    """初始化RabbitMQ连接、通道和队列"""
    global global_connection, global_channel, global_queue
    # 创建健壮连接（自动重连）
    global_connection = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost/"
    )
    # 创建通道
    global_channel = await global_connection.channel()
    # 声明持久化队列
    global_queue = await global_channel.declare_queue("my_target_queue", durable=True)

async def process_message(message: aio_pika.IncomingMessage):
    """消息处理函数"""
    async with message.process():
        print(f"收到消息: {message.body.decode()}")

# 程序入口
async def main():
    # 先初始化RabbitMQ资源
    await init_rabbitmq()
    # 开始消费队列
    await global_queue.consume(process_message)
    # 保持程序运行（按需调整）
    await global_connection.close()

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

如果是基于异步框架（如FastAPI），可以利用框架的启动/关闭事件完成初始化：

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup():
    await init_rabbitmq()

@app.on_event("shutdown")
async def shutdown():
    await global_connection.close()

方案二：单例类封装连接逻辑 #

用单例类封装RabbitMQ的连接、通道和队列管理，避免全局变量的混乱，更适合大型项目。

示例代码：

import aio_pika
from aio_pika import Connection, Channel, Queue

class RabbitMQClient:
    _instance = None
    connection: Connection | None = None
    channel: Channel | None = None
    queue: Queue | None = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def initialize(self, amqp_url: str, queue_name: str):
        """初始化连接、通道和队列"""
        if not self.connection:
            self.connection = await aio_pika.connect_robust(amqp_url)
            self.channel = await self.connection.channel()
            self.queue = await self.channel.declare_queue(queue_name, durable=True)

    async def close(self):
        """关闭连接"""
        if self.connection:
            await self.connection.close()

# 使用示例
async def main():
    client = RabbitMQClient()
    await client.initialize("amqp://guest:guest@localhost/", "my_target_queue")
    await client.queue.consume(process_message)
    await client.close()

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

关键注意事项 #

• 必须在异步函数上下文中执行连接初始化，不能在模块顶层直接使用await
• 优先使用connect_robust创建连接，它会自动处理重连逻辑，提升可靠性
• 程序退出时务必关闭连接，避免资源泄漏

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