如何将RabbitMQ连接设为全局并解决AttributeError报错？

你碰到的AttributeError: 'coroutine' object has no attribute 'channel'是因为connect()是异步函数——直接在全局作用域调用它不会实际建立连接，只会返回一个协程对象，而非可用的RabbitMQ连接实例，自然找不到channel方法。

要实现全局复用连接，得用异步初始化的方式，步骤如下：

1. 声明全局连接变量 #

先在全局作用域定义一个变量，初始化为None，用来存储后续建立的连接：

import asyncio
from aio_pika import connect, Message
import json

# 全局RabbitMQ连接变量
rabbitmq_connection = None

2. 写异步初始化函数 #

单独写一个异步函数，专门负责建立RabbitMQ连接，在程序启动时调用它：

async def init_rabbitmq():
    global rabbitmq_connection
    # 用await执行异步连接操作，得到实际的连接实例
    rabbitmq_connection = await connect("amqp://guest:guest@localhost/")

3. 修改消息发送函数 #

直接复用全局连接，不用每次创建和关闭连接：

async def send_rabbitmq(msg = {}):
    # 先检查连接是否已初始化
    if not rabbitmq_connection:
        raise RuntimeError("RabbitMQ连接未初始化，请先调用init_rabbitmq()")
    
    channel = await rabbitmq_connection.channel()
    await channel.default_exchange.publish(
        Message(json.dumps(msg.dict()).encode("utf-8")),
        routing_key = "fastapi_task"
    )
    # 注意：不要在这里关闭连接，全局连接要复用，只在程序退出时关闭

4. 程序启动时初始化连接 #

如果是FastAPI这类异步框架，可以利用启动事件自动初始化：

from fastapi import FastAPI

app = FastAPI()

# 启动时初始化RabbitMQ连接
@app.on_event("startup")
async def startup_event():
    await init_rabbitmq()

# 关闭时释放连接
@app.on_event("shutdown")
async def shutdown_event():
    if rabbitmq_connection:
        await rabbitmq_connection.close()

注意事项 #

• 全局连接必须在所有异步任务执行前完成初始化，不能在全局代码里直接await connect()，因为全局代码是同步执行的，无法处理异步操作
• 复用连接时，不要每次发送消息都关闭连接，仅在程序退出时统一关闭
• 可以给发送函数加个重连逻辑，避免连接意外断开后无法发送消息

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