Python 并发程序中的常见坑点

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 循环里。