Python asyncio并发编程

你写完一个 async def 函数，跑起来却发现比同步还慢，或者并发一开就撞上限流、协程报错还看不到堆栈。这篇不解释协程调度原理，只讲工程上怎么把 asyncio 用对：怎么发任务、怎么卡住并发数、怎么设超时和重试、怎么打日志定位卡在哪一步，最后给一份能照着走的排查顺序。

你写完一个 async def 函数，跑起来却发现比同步还慢，或者并发一开就撞上限流、协程报错还看不到堆栈。这篇不解释协程调度原理，只讲工程上怎么把 asyncio 用对：怎么发任务、怎么卡住并发数、怎么设超时和重试、怎么打日志定位卡在哪一步，最后给一份能照着走的排查顺序。

先明确一件事：asyncio 的价值在于把等待时间叠起来，不在于让代码变快。只要你的任务里有真实 IO 等待，它就有用；只要里面有 CPU 密集计算，它就基本帮不上忙。全文围绕四个动作展开：定任务模型、关并发、兜超时重试、埋日志排查。

先把任务模型定下来：谁并发、并发多少个

很多人第一步就写错：把所有请求塞进一个 gather 一次性铺开，看着很爽，结果对方限流、本地文件句柄打满、内存里堆着几万条未处理结果。发文之前先想清楚总量是已知还是流式，再决定用哪种写法。

• 并发上限的来源先搞清楚：是本地限制、接口方的限额、还是下游数据库连接数，三者取最小那个
• 先用小规模跑一轮把单请求延迟基线量出来，再决定并发数，别拍脑袋；压测工具和基线口径可以到 /tools/speed-test 先测一遍
• 批大小、每批之间的间隔写成配置项，不要散落在代码里，调参时不用改逻辑
• 任务里混了 CPU 密集计算时，asyncio 帮不上忙，要么换多进程，要么把计算丢到线程池

用 Semaphore 和队列把并发关进笼子

并发数不设上限，本质上是把压力转嫁给对方。Semaphore 是最直接的刹车，但位置很关键：它要包在真正发请求的那一层，而不是包在外层循环里。包错地方，信号量就是个装饰。

1. 确定并发数按下游限额留出余量，先设一个小值跑通，再按成功率往上加；加到失败率抬头就退回来。
2. 加信号量async with sem: 包住单次请求调用，确保每次真正发起请求前先拿到槽位。
3. 处理未知总量任务数不清楚就用 asyncio.Queue 加固定数量 worker，消费完发哨兵让 worker 退出。
4. 验证效果打点记录同时在跑的协程数，确认它没有超过你设的上限，超了就说明信号量位置放错了。

还有一个常见坑是嵌套扇出：外层一个协程拿到信号量后，内部又开一批子协程，子协程没被限制，总数就失控。子任务数量要乘进总预算里算。

超时、重试、降级三件事必须一起写

只加重试不加超时，等于给自己埋雷：一个卡住的请求会一直占着槽位，整批任务跟着一起等。超时、重试、降级要当成一组写，缺一个都会在异常情况下暴露问题。

超时

用 asyncio.wait_for 或 asyncio.timeout 给每个请求兜底，值按接口的 P99 延迟往上留余量，别用框架默认值

退避重试

只对网络类和限流类错误做，指数退避加随机抖动，重试次数和总耗时都要有上限

降级

提前定好备用模型、备用地址、返回缓存，还是标记失败丢进待重跑队列

• 参数错、鉴权错重试没有意义，几分钟后再试还是同样的结果，直接失败更快
• 重试次数乘以超时值，就是一条任务的最坏耗时，这个数字要乘以并发数估算整体时长
• 重试不要把总耗时堆成串行等待，多个失败任务同时退避时，整体时间会被最慢那个拖住
• 并发和退避之间的取舍，可以到 /guides/429-concurrency-backoff 看当时的处理思路

日志和排查：协程卡住时怎么找到它

协程报错堆栈经常断在 await 那一行，看不出到底卡在哪一步。解法不是去读源码，而是把每条任务的关键节点打出来：任务 ID、开始时间、耗时、状态、拿到的并发槽位号。有了这几个字段，按时间线一拼就能看出谁在等谁。

用 asyncio.gather(return_exceptions=True) 收集异常，别让单个任务的失败静默消失；批量跑完统计一下成功率和错误类型分布，比逐条看日志高效得多。

• 先确认并发数实际生效：打点看同时在跑的协程数是否接近设定的上限
• 再看单请求耗时分布：是普遍慢还是个别慢，个别慢优先查超时和重试配置
• 然后看超时与重试叠加后有没有变成新的串行等待
• 最后看错误类型：是本地参数问题还是下游限流问题，两边处理方式完全不同