




asyncio.run() 只能在顶层脚本入口调用,不可在已运行事件循环(如Jupyter、FastAPI)中重复使用;需用create_task()或await替代;协程必须显式await,否则不执行;共享状态须用asyncio.Lock()保护;CPU密集任务须用run_in_executor()或to_thread()卸载。
这是新手最常遇到的 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+ 已不推荐,且在嵌套场景下行为更难预测常见现象是程序“看似跑完了”,但异步任务根本没执行——比如忘了在 asyncio.sleep(1) 前加 await,或者把 fetch_data()(返回协程对象)直接传给 print() 而不是 await fetch_data()。
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 一律视为不安全asyncio 不是万能加速器。如果你在协程里写了个 for i in range(10**7): total += i,整个事件循环就卡死了——因为没 await,就没有让出控制权的机会。
这种问题在本地测试时不易察觉(小数据量快),一上生产就暴露

loop.run_in_executor() 丢给线程池或进程池asyncio.to_thread()(Python 3.9+)替代手写 executor 调用,更简洁time.sleep(),它会彻底阻塞;必须用 await asyncio.sleep()
真正难处理的从来不是“怎么写 async”,而是判断哪些操作必须 await、哪些必须 offload、哪些压根不该放进协程里。边界模糊的地方,往往藏在第三方库文档的角落,或你自己的 for 循环里。