asyncio.run() 只能在顶层脚本入口调用,不可在已运行事件循环(如Jupyter、FastAPI)中重复使用;需用create_task()或await替代;协程必须显式await,否则不执行;共享状态须用asyncio.Lock()保护;CPU密集任务须用run_in_executor()或to_thread()卸载。

asyncio.run() 不能在已运行的事件循环中调用
这是新手最常遇到的 RuntimeError: asyncio.run() cannot be called from a running event loop。根本原因不是代码写错了,而是你在 Jupyter、IPython 或已启动 asyncio 的上下文(比如 FastAPI 启动后)里又调用了 asyncio.run()。
解决方法很简单:只在顶层脚本入口用一次 asyncio.run();在交互环境或已有事件循环中,改用 asyncio.create_task() 或 await 直接执行协程。
- ✅ 正确:脚本最外层
asyncio.run(main()) - ❌ 错误:在
async def handler()里再写asyncio.run(another_coro()) - ⚠️ 注意:
asyncio.get_event_loop().run_until_complete()在 Python 3.10+ 已不推荐,且在嵌套场景下行为更难预测
await 一个普通函数或未 await 的协程对象
常见现象是程序“看似跑完了”,但异步任务根本没执行——比如忘了在 asyncio.sleep(1) 前加 await,或者把 fetch_data()(返回协程对象)直接传给 print() 而不是 await fetch_data()。
Python 不会报错,只会打印类似 ,而后续逻辑可能因变量类型错误崩溃。
立即学习“Python免费学习笔记(深入)”;
- ✅ 检查所有调用:确认函数是否带
async声明,如果是,必须await - ✅ 用
inspect.iscoroutine()或inspect.iscoroutinefunction()在调试时验证返回值类型 - ⚠️ 特别注意第三方库:有些函数名像异步(如
aiohttp.ClientSession.get),但返回的是ClientResponse对象,真正要 await 的是它的.text()或.json()方法
共享状态未加锁导致竞态条件
很多人以为 “async 就是线程安全的”,结果在多个协程里同时修改一个全局列表或字典,出现数据丢失或索引错乱。asyncio 是单线程并发,但协程切换点(await)就是竞态窗口。
典型错误:多个协程都执行 results.append(data),但 append 不是原子操作(先读长度、再写入、再更新长度)。
- ✅ 用
asyncio.Lock()包裹临界区,不是threading.Lock - ✅ 更推荐无状态设计:每个协程生成独立结果,最后用
asyncio.gather()收集,避免共享可变状态 - ⚠️
asyncio.Queue是线程/协程安全的,适合生产者-消费者模式;但直接读写普通 dict/list 一律视为不安全
长时间 CPU 密集型操作阻塞整个事件循环
asyncio 不是万能加速器。如果你在协程里写了个 for i in range(10**7): total += i,整个事件循环就卡死了——因为没 await,就没有让出控制权的机会。
这种问题在本地测试时不易察觉(小数据量快),一上生产就暴露:HTTP 超时、心跳断连、其他协程饿死。
- ✅ CPU 密集任务必须移出事件循环:用
loop.run_in_executor()丢给线程池或进程池 - ✅ 用
asyncio.to_thread()(Python 3.9+)替代手写 executor 调用,更简洁 - ⚠️ 不要用
time.sleep(),它会彻底阻塞;必须用await asyncio.sleep()
真正难处理的从来不是“怎么写 async”,而是判断哪些操作必须 await、哪些必须 offload、哪些压根不该放进协程里。边界模糊的地方,往往藏在第三方库文档的角落,或你自己的 for 循环里。










