asyncio集成aiohttp报event loop already running的实现方案咨询

错误原因 #

web.run_app()为同步阻塞接口，内部会主动启动新的事件循环并调用run_forever()，与你已经通过asyncio.run()启动的现有事件循环冲突，因此触发报错。

正确实现方案 #

无需额外启动线程，直接改用aiohttp低阶挂载接口AppRunner，将web服务挂载到现有事件循环即可，完整修改代码如下：

import asyncio
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return web.Response(text=text)

async def start_api_server():
    print("api server starting")
    app = web.Application()
    app.add_routes([web.get('/', handle),
                    web.get('/{name}', handle)])
    # 改用AppRunner挂载服务，不单独启动事件循环
    runner = web.AppRunner(app)
    await runner.setup()
    # 配置监听地址端口，可根据需求修改
    site = web.TCPSite(runner, '0.0.0.0', 8080)
    await site.start()
    print("api server listening on 0.0.0.0:8080")
    # 永久运行保持服务存活
    while True:
        await asyncio.sleep(3600)

async def action():
    print("action task running")
    # 原有业务逻辑保持不变
    await asyncio.sleep(10000)

async def main():
    # 两个任务共享同一个事件循环并发运行
    await asyncio.gather(start_api_server(), action())

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

改动说明 #

• 替换web.run_app()为AppRunner + TCPSite组合接口，该方式仅将web服务注册到当前正在运行的事件循环，不会尝试启动新的事件循环
• 服务启动后通过长休眠保持进程存活，和原有业务逻辑任务通过asyncio.gather并发调度，互不阻塞
• 若需要查询内部状态，直接在接口处理函数中读取你要暴露的全局/共享变量即可，不需要额外的跨线程通信逻辑

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