uvicorn环境下asyncio.run报错及Pub/Sub订阅后台启动咨询

1. 为什么会出现RuntimeError？ #

你的判断完全正确——asyncio.run()的用法确实有误，而且Uvicorn已经在运行一个asyncio事件循环了：

• asyncio.run()的核心作用是创建全新事件循环并运行协程，执行结束后还会关闭这个循环。
• Uvicorn作为ASGI服务器，启动时就会初始化并持续运行一个事件循环，所有接口请求、异步任务都在这个循环里处理。
• 在已有运行中的事件循环里调用asyncio.run()，必然触发“不能从运行中循环调用asyncio.run()”的冲突错误。

2. 如何在后台启动Pub/Sub订阅？ #

无需额外创建新循环，直接利用Uvicorn已有的事件循环，把订阅协程包装成后台任务即可，这样不会阻塞main.py的执行，控制权会立刻返回。

方案一：利用框架生命周期事件（推荐，如FastAPI/Starlette） #

如果你的应用用了FastAPI这类支持生命周期事件的框架，直接在启动事件里创建后台任务：

import asyncio
from fastapi import FastAPI

app = FastAPI()

# 你的Pub/Sub订阅协程
async def create_bill_subscriber():
    # 初始化订阅者逻辑
    subscriber = # 这里写入你的Pub/Sub订阅者初始化代码
    async with subscriber:
        async for message in subscriber:
            # 处理消息的逻辑
            print(f"收到消息: {message.data}")
            message.ack()

# 在应用启动时触发后台任务
@app.on_event("startup")
async def start_subscriber():
    # 将订阅协程加入当前事件循环，后台异步执行
    asyncio.create_task(create_bill_subscriber())

方案二：直接获取当前事件循环添加任务 #

如果没有用框架的生命周期事件，确保在Uvicorn事件循环启动后执行以下代码：

import asyncio

# 你的Pub/Sub订阅协程
async def create_bill_subscriber():
    # ... 订阅逻辑 ...

# 适合用代码直接启动Uvicorn的场景
if __name__ == "__main__":
    import uvicorn
    loop = asyncio.get_event_loop()
    loop.create_task(create_bill_subscriber())
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键注意事项 #

• asyncio.create_task()会把协程放到当前事件循环的任务队列中，后台异步执行，不会阻塞主线程后续逻辑。
• 不要在脚本最外层直接调用asyncio.create_task()，必须确保代码在Uvicorn的事件循环启动后执行（比如启动事件、代码启动Uvicorn前）。

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