asyncio.run() 并非万能主入口,仅适用于无运行中事件循环的简单脚本;它会强制新建并关闭事件循环,导致后台任务被取消、全局异步资源失效,且不支持嵌入已有环境或复用循环。

asyncio.run() 是不是万能的主入口
绝大多数简单脚本用 asyncio.run() 没问题,但它只在“没有运行中的事件循环”时才安全。一旦你在 Jupyter、Django shell、或已手动调用 asyncio.new_event_loop() 的环境里再调用它,就会报 RuntimeError: asyncio.run() cannot be called from a running event loop。
常见踩坑场景:
- Jupyter Notebook 里反复执行
asyncio.run()单元格(第二次就崩) - Flask 或 FastAPI 的同步路由里直接调异步函数并用
run()包裹 - 测试代码中多个
asyncio.run()调用没做隔离,导致循环被重复关闭又重建
为什么 asyncio.run() 会自动关闭 EventLoop
它内部做了三件事:新建 loop → 调用你的协程 → 调用 loop.close()。这个关闭是强制的,哪怕协程里 spawn 了后台任务(比如 asyncio.create_task()),只要主协程返回,loop 就关,后台任务直接被 cancel。
所以别指望靠 asyncio.run() 做长时后台服务——它不是设计来托管长期运行的事件循环的。
- 后台任务没 await 完就被 kill,不会抛异常,但你会看到
Task was destroyed but it is pending!警告 - 如果协程里用了
async with aiohttp.ClientSession(),session 会在 loop 关闭前被正确退出;但如果 session 被存成全局变量,下次再用就失效 -
asyncio.run()不接受debug=True以外的 loop 参数,没法传自定义策略或已配置好的 loop
什么时候该换掉 asyncio.run()
当你需要控制生命周期、复用 loop、或嵌入已有运行环境时,就得绕过 asyncio.run()。
- 在主线程已有 loop(如 PyQt/TKinter 的异步集成),改用
asyncio.get_event_loop().run_until_complete() - 想让多个协程共享同一个 loop(比如同时跑 HTTP client 和 WebSocket server),手动创建并保持 loop 实例
- 写 CLI 工具且需支持子命令异步执行,避免每次子命令都新建+关闭 loop,影响性能和资源清理
- 使用
trio或curio等替代库时,asyncio.run()根本不适用
一个安全的手动 loop 启动模板
如果你必须自己管理 loop(比如兼容 Python 3.6 或规避 run() 的限制),下面是最简可行写法:
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(main())
finally:
loop.close()
注意两点:
- Python ≥ 3.7 推荐用
asyncio.run(),除非你明确知道为什么不用 -
loop.close()后不能再用这个 loop,也不能再调set_event_loop()绑定已关闭的 loop - Windows 上默认策略是
ProactorEventLoop,某些低层操作(如 subprocess)可能要求切换回SelectorEventLoop
真正麻烦的从来不是怎么启,而是启完之后谁负责关、什么时候关、关之前有没有漏掉 await 的 cleanup —— 这些细节藏在业务逻辑深处,而不是 API 文档第一行。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











