调试、性能和容易写坏的地方
asyncio 程序的 bug 常常不在语法上。更常见的问题是：某个协程没有被等待，某个同步调用堵住了事件循环，或者后台 Task 抛异常没人取。这一章给你一套排查顺序。
打开 debug 模式
可以用几种方式打开：
bash
PYTHONASYNCIODEBUG=1 python3 app.py
python3 -X dev app.py或在代码里：
python
asyncio.run(main(), debug=True)再把日志和资源警告打开：
python
import logging
import warnings
logging.basicConfig(level=logging.DEBUG)
warnings.simplefilter("default", ResourceWarning)debug 模式会帮助发现：
- 非线程安全 API 从错误线程调用。
- selector I/O 操作耗时过长。
- callback 执行时间超过慢回调阈值。
- coroutine 创建了但没有被等待。
- Future/Task 异常从未被取走。
忘记 await
坏写法：
python
async def save():
...
save()这只创建 coroutine object，不会执行函数体。debug 模式下你会看到相关警告。
修正：
python
await save()如果需要并发执行：
python
task = asyncio.create_task(save())
await task阻塞事件循环
坏写法：
python
async def handler():
time.sleep(1)time.sleep() 会阻塞线程，事件循环没有机会运行别的 Task。改成：
python
await asyncio.sleep(1)如果确实要调用同步阻塞函数：
python
result = await asyncio.to_thread(blocking_function, arg)也可以直接用线程池：
python
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, blocking_function, arg)这适合包一段绕不开的同步 I/O，不适合把大量 CPU 计算都塞进默认线程池。
get_event_loop 的历史包袱
在协程和 callback 里取正在运行的 loop，优先：
python
asyncio.get_running_loop()get_event_loop() 的行为和当前线程、policy、历史版本相关，容易让代码含糊。Python 3.14 文档已经记录它在没有当前事件循环时会抛 RuntimeError；3.15 开发文档还提示 policy system 未来会移除。
常见写坏方式
| 写法 | 症状 | 更好的写法 |
|---|---|---|
async def 里调用 time.sleep() | 所有请求卡住 | await asyncio.sleep() 或 to_thread() |
循环里立即 await 每个任务 | 并发退化成串行 | 先 create_task，再统一等待 |
| 后台 Task 不保存引用 | 异常丢失、生命周期失控 | TaskGroup 或集合管理 |
| 捕获宽泛异常吞掉取消 | 超时和关闭不可靠 | 对 CancelledError 清理后重新抛出 |
| 无限创建 Task 不限流 | 内存暴涨、远端被打爆 | Queue(maxsize) 或 Semaphore |
忘记 writer.wait_closed() | 连接关闭不完整 | writer.close(); await writer.wait_closed() |
性能直觉
asyncio 的性能收益来自减少等待浪费，不是让单个 Python 函数跑得更快。
排查性能问题时，可以按这个顺序来：
- 先确认慢点是否是 I/O 等待。
- 给外部调用加超时。
- 用
Semaphore或Queue(maxsize)做限流。 - 检查是否有同步阻塞调用。
- 打开 debug 模式看慢 callback。
- 必要时把 CPU 密集部分移出事件循环。
小练习
运行：
bash
python3 examples/asyncio_demos/11_debug_antipatterns.py你会看到未等待 coroutine 和慢 callback 相关提示。然后把 time.sleep(0.2) 改成 await asyncio.sleep(0.2)，再运行一次比较输出。