← 中文资源

简体中文版本

Python — RuntimeError: Event loop is closed

把 asyncio 事件循环关闭错误追踪到重复 asyncio.run、已关闭客户端或测试夹具作用域,修复生命周期而不是捕获错误。

阅读英文原文

一个循环,一个生命周期

asyncio.run() 创建新事件循环、运行协程并关闭循环。一个进程里调用两次没问题,前提是第一次运行的任何对象都不再存活。异步客户端、会话和后台任务往往会存活,并在之后尝试使用已关闭的循环。

async def main():
    async with httpx.AsyncClient() as client:
        await fetch_all(client)

asyncio.run(main())

客户端应在 main 内创建,而不是在模块导入时。返回前要 await 或取消后台任务。不要保留循环引用并跨多次 asyncio.run 复用。

测试需要匹配作用域

在一个循环中创建客户端、在另一个循环中关闭的测试夹具会产生此错误。使用 pytest-asyncio 的异步夹具,让事件循环作用域与资源一致,并在同一循环中关闭客户端。同一进程跑两次测试可以暴露泄漏的循环状态。

关闭待处理工作

关闭时取消剩余任务并 await 其取消。asyncio.all_tasks() 显示仍在运行的任务。待处理时被销毁的任务可能记录警告并丢失工作。目标是干净关闭循环,而不是用异常处理器掩盖症状。

用调试模式验证

使用 PYTHONASYNCIODEBUG=1 和 warnings-as-errors 运行。确认没有残留待处理任务,并反复运行测试套件。如果错误仍在,找出在循环外创建的对象;那就是生命周期不匹配。