asyncio.run() 会自动创建、运行并关闭事件循环,手动调用 loop.close() 必然触发 RuntimeError;异步生成器未耗尽、跨线程误用 loop、pytest-asyncio fixture 清理顺序不当也会引发该错误。

asyncio.run() 调用后还手动调用 loop.close()?
这是最常见触发 RuntimeError: Event loop is closed 的原因。Python 3.7+ 的 asyncio.run() 内部会自动创建、运行、关闭事件循环,你再显式关一次,就直接报错。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 只用
asyncio.run(main())启动主协程,别碰get_event_loop()或close() - 如果必须复用循环(比如测试中多次调用),改用
asyncio.new_event_loop()+ 手动run_until_complete()+close(),但要确保只关一次 - 检查是否在
atexit或__del__里偷偷调了loop.close()——asyncio.run()结束后,loop 对象已失效,此时 close 就是无效操作加报错
异步生成器(async def + yield)没被完全消费完
未遍历完的异步生成器会在垃圾回收时尝试清理自身,但若此时事件循环已关闭(比如 asyncio.run() 已退出),就会抛出 RuntimeError,且常伴随 async_generator_athrow(): object is not an async generator 这类次级错误。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 所有
async for必须完整执行,或用try/finally确保agen.aclose()被调用(注意:不是agen.close()) - 避免把异步生成器返回给上层后“丢着不管”,尤其在装饰器、上下文管理器或缓存逻辑中
- 调试时可在生成器末尾加
print("agen done"),确认是否真走到了 yield 结束之后
跨线程调用 loop.create_task() 或 loop.run_in_executor()
事件循环默认绑定到创建它的线程,其他线程直接调用 create_task() 会因找不到当前 loop 而报错;更隐蔽的是:某些线程里用 asyncio.get_event_loop() 拿到的可能是已关闭的 loop(比如主线程的 loop 已随 asyncio.run() 结束)。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 跨线程提交任务,必须用
loop.call_soon_threadsafe()包裹create_task(),不能直接调 - 子线程中不要依赖
get_event_loop(),应由主线程把 loop 实例传入,或用asyncio.run_coroutine_threadsafe(coro, loop) - 检查日志中是否出现
There is no current event loop in thread 'xxx'—— 这往往是后续Event loop is closed的前兆
使用 pytest-asyncio 时 fixture 清理顺序错乱
当 fixture 声明为 scope="function" 且含 async def,pytest-asyncio 默认在每个 test 后 await cleanup。但如果 cleanup 协程里用了已关闭的 loop(比如它依赖全局 loop,而该 loop 在 session 级 teardown 中被关了),就会爆 RuntimeError。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 所有 async fixture 的 cleanup 都应封装在
async with或try/finally中,并捕获RuntimeError(仅针对 loop 关闭场景,不掩盖其他错误) - 避免在
session或module级 fixture 中启动/关闭 loop;改用function级 +asyncio.new_event_loop()隔离 - 升级
pytest-asyncio >= 0.21.0,旧版本对 loop 生命周期管理较粗放,容易在 teardown 阶段误用已关闭 loop
真正麻烦的是异步资源生命周期和事件循环生命周期的耦合点 —— 它们不在同一个抽象层,但错误往往发生在交界处。盯住谁关 loop、谁还在用 loop、谁在哪个线程里拿 loop,比看堆栈更管用。









