1. 不要在协程中调用同步阻塞函数
import time
async def bad():
time.sleep(1) # 错误！将阻塞整个事件循环
async def good():
await asyncio.sleep(1) # 正确！主动让出控制权
async def acceptable():
await asyncio.to_thread(time.sleep, 1) # 可行！让线程池去阻塞2. 始终使用 asyncio.run 作为入口
# 错误：手动管理事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
task = loop.create_task(main())
loop.run_forever()
# 正确：asyncio.run 自动创建和关闭事件循环
asyncio.run(main())3. 小心协程对象未被 await
async def main():
# 错误：创建了协程但未 await，协程永远不会执行
fetch("url", 1)
# 正确
await fetch("url", 1)
# 或通过 gather
await asyncio.gather(fetch("url1", 1), fetch("url2", 2))4. gather 异常处理
gather 默认任何协程异常都会立即传播，其他协程不会取消，但结果丢失
async def main():
# 方式一：使用 return_exceptions=True
results = await asyncio.gather(
risky_task(),
safe_task(),
return_exceptions=True,
)
for r in results:
if isinstance(r, Exception):
print(f"某个任务失败: {r}")
# 方式二：使用 TaskGroup（Python 3.11+）
# 任一异常会取消组内所有任务5. 使用 debug 模式
asyncio 的 debug 模式可以帮助你发现异步代码中的常见问题，比如协程阻塞事件循环、忘记 await、回调执行时间过长等。
# 开启 asyncio 调试模式
asyncio.run(main(), debug=True)
# 或通过环境变量
# PYTHONASYNCIODEBUG=1 python script.py检测长时间阻塞的协程
debug 模式下，事件循环会监控每个协程的执行时间。如果某个协程执行超过 0.1 秒（默认阈值），会在 stderr 输出警告：
import time
import asyncio
async def blocking_coroutine():
"""模拟一个协程内部做了同步阻塞操作"""
print("开始阻塞操作...")
time.sleep(0.2) # 同步阻塞，会阻塞整个事件循环
print("阻塞操作结束")
async def main():
await blocking_coroutine()
asyncio.run(main(), debug=True)输出类似：
开始阻塞操作...
阻塞操作结束
Executor <TaskInfo name='Task-1' ...> running at (...)
blocking_coroutine at demo.py:12
main at demo.py:18
...time.sleep(0.2) 是同步阻塞，但 debug 模式下检测到协程在同一个位置停留超过 0.1 秒，会打印出执行栈信息，精确定位阻塞的代码行。
检测未 await 的协程对象
忘记 await 协程是新手最容易犯的错误，debug 模式会检测到协程对象被创建但从未被迭代：
import asyncio
async def fetch_data(url: str) -> str:
await asyncio.sleep(0.5)
return f"{url} 数据"
async def main():
# 忘记 await，协程对象永远不会执行
fetch_data("https://example.com")
await asyncio.sleep(1)
asyncio.run(main(), debug=True)
输出类似：
Coroutine 'fetch_data' was never awaited (at demo.py:12)这个警告在你忘记 await 时非常有用，避免协程”静默丢失”。
自定义慢操作阈值
通过 loop.slow_callback_duration 调整检测阈值：
async def main():
loop = asyncio.get_running_loop()
loop.slow_callback_duration = 0.5 # 改为 0.5 秒才报警
def acceptable_callback():
time.sleep(0.3) # 0.3 秒，低于自定义阈值，不会报警
loop.call_later(0.1, acceptable_callback)
await asyncio.sleep(0.5)
asyncio.run(main(), debug=True)6. 避免全局事件循环
# 错误：在模块级别获取事件循环
loop = asyncio.get_event_loop() # 可能获取到错误的事件循环
# 正确：在协程内部获取当前事件循环
async def my_func():
loop = asyncio.get_running_loop()7. CancelledError 的正确处理
async def cleanup():
try:
await long_running_task()
except asyncio.CancelledError:
# 务必完成清理后再重新抛出
await release_resources()
raise # 必须重新抛出
