如何解决Python异步程序中的Event loop is closed错误？

asyncio.run() 会创建并显式关闭事件循环，之后 asyncio.get_event_loop() 返回已关闭的 loop，导致 RuntimeError。正确做法是每个异步入口都用 asyncio.run() 包裹，避免混用 get_event_loop()。

为什么 asyncio.run() 后再调用 asyncio.get_event_loop() 会报 “Event loop is closed”

因为 asyncio.run() 内部会创建新 event loop、运行协程、然后**显式关闭并丢弃该 loop**。之后再调用 asyncio.get_event_loop()，拿到的仍是那个已被关闭的 loop 实例（不是新 loop），调用 run_until_complete() 等方法就会触发 RuntimeError: Event loop is closed。

常见于写完一个 asyncio.run(main()) 后，又在 REPL 或脚本后续手动尝试：

asyncio.get_event_loop().run_until_complete(some_coro())

这几乎必错。

• ✅ 正确做法：每个顶层异步入口都用 asyncio.run() 包一层，不要混用 get_event_loop()
• ⚠️ 注意：asyncio.run() 是“一次性”的，不能复用；它不等价于“启动一个长期运行的 loop”
• ❌ 不要试图在 asyncio.run() 之后保留或重用 loop 引用

在已关闭的 loop 上调用 loop.create_task() 或 loop.run_in_executor() 怎么办

本质还是 loop 已关闭，但错误可能更隐蔽——比如你在类的 __del__、atexit 回调、或信号处理器里试图调度任务，而此时主 loop 早已退出。

立即学习“Python免费学习笔记（深入）”；

典型场景：

• 使用 atexit.register() 清理异步资源（如关闭 aiohttp.ClientSession）
• 对象析构时想 await 一个 cleanup 协程
• 收到 SIGINT 后尝试 graceful shutdown

解决思路不是“重启 loop”，而是绕过 loop：

• 用 asyncio.run_coroutine_threadsafe(coro, loop) —— 但前提是 loop 还在运行且未关闭（适合后台线程中提交任务）
• 对清理逻辑做降级：同步方式关闭（如直接调用 session.connector._close()，不推荐但有时是唯一选择）
• 更健壮的做法：把 cleanup 注册到正在运行的 loop 中，例如在 main 协程末尾 await cleanup()，或用 asyncio.shield() 包住关键清理任务

多线程环境下误用主线程 loop 导致 closed 错误

当从子线程访问 asyncio.get_event_loop()，Python 默认返回主线程的 loop（如果存在），但该 loop 很可能已在主线程中被 asyncio.run() 关闭。子线程不能直接复用主线程的已关闭 loop。

正确做法是：子线程中必须自己创建并管理 loop：

import asyncio<br>def worker():<br>    loop = asyncio.new_event_loop()<br>    asyncio.set_event_loop(loop)<br>    loop.run_until_complete(my_coro())<br>    loop.close()

• ⚠️ 切记：每个线程最多一个 loop，且需手动 set_event_loop() 绑定
• ❌ 不要用 asyncio.get_event_loop() 在非主线程中获取 loop —— 它不保证返回当前线程的 loop（Python 3.12+ 默认行为已变更，但兼容性差）
• ✅ 更安全的选择：始终用 asyncio.new_event_loop() + set_event_loop() 显式初始化

使用 nest_asyncio 的真实适用边界

nest_asyncio 的作用是给已关闭的 loop “打补丁”，让它能重新接受新任务 —— 这仅适用于 Jupyter、IPython、某些测试框架等**无法控制 loop 生命周期的嵌入环境**。

它不是通用解法：

• ✅ 适合：Jupyter notebook 里反复执行 await coro，或 pytest-asyncio 测试中多次调用 asyncio.run()
• ❌ 不适合：生产服务、CLI 工具、长期运行的后台任务 —— 它掩盖了 loop 管理混乱的问题，且在 Python 3.12+ 下行为不稳定
• ⚠️ 注意：必须在 import 后立即调用 nest_asyncio.apply()，且只能调用一次；否则可能引发 loop 状态错乱

真正需要关注的，不是怎么让 closed loop “复活”，而是厘清谁该创建 loop、谁该关闭它、以及任务是否真的需要在那个 loop 上跑。loop 关闭不是 bug，是设计预期；错误往往来自跨生命周期误用。