Python 异步任务里的并发控制：Semaphore、超时与指数退避

一个常见的错误写法

需要并发请求 1000 个 URL 时，最直觉的写法是：

tasks = [fetch(u) for u in urls]

results = await asyncio.gather(*tasks)

这段代码在本地跑得好好的，上线就出问题：1000 个协程同时发起连接，本地文件描述符耗尽直接 Too many open files；对端看到瞬时 1000 QPS，轻则限流，重则拉黑。

gather 本身没有并发上限，它只是把协程全部铺开。

用 Semaphore 限流

asyncio.Semaphore 是最直接的解法：

import asyncio

sem = asyncio.Semaphore(20) # 同时最多 20 个在跑

async def fetch(url):

    async with sem:

        async with session.get(url) as resp:

            return await resp.text()

注意 async with sem 必须包住整个请求过程，而不是只包住建连。如果只包住 session.get()，await resp.text() 读响应体时信号量已经释放，实际并发还是会超。这个坑很常见，因为 async with 的作用域很容易写窄。

限流之外还要限速

Semaphore 控制的是「同时在跑多少个」，不是「每秒发多少个」。20 个并发、每个请求 50ms，实际就是 400 QPS。如果对端限制的是 QPS 而不是并发数，还需要一个令牌桶：

class RateLimiter:

    def __init__(self, qps: float):

        self.interval = 1 / qps

        self._next = 0.0

        self._lock = asyncio.Lock()

    async def acquire(self):

        async with self._lock:

            loop = asyncio.get_running_loop()

            now = loop.time()

            wait = max(0.0, self._next - now)

            self._next = max(now, self._next) + self.interval

        if wait:

            await asyncio.sleep(wait)

这里的关键是「占位」在锁内完成、sleep 在锁外执行。如果把 sleep 放在锁里，所有协程会串行等待，限速器退化成串行执行器。

每个请求都必须有超时

没有超时的 await 是生产事故的温床。一个卡住的连接会一直占着信号量，20 个信号量被占满后，整个任务就僵死了——而且不会报错，只会一直「正在运行」。

async with asyncio.timeout(10): # Python 3.11+

    async with sem:

        ...

3.11 之前用 asyncio.wait_for(coro, 10)。要注意 wait_for 取消协程后，底层的连接不一定会立即关闭，用 aiohttp 时建议显式设置 ClientTimeout(total=10, connect=3)，让库自己管住连接生命周期。

重试要退避，而且要区分错误类型

不是所有失败都值得重试。404、401 重试一百次也是同样的结果，只会浪费额度；429、500、连接重置才需要退避重试。

async def fetch_with_retry(url, retries=3):

    for attempt in range(retries + 1):

        try:

            async with sem:

                async with asyncio.timeout(10):

                    async with session.get(url) as resp:

                        if resp.status in (429, 500, 502, 503, 504):

                            raise RetryableError(resp.status)

                        resp.raise_for_status()

                        return await resp.text()

        except RetryableError as e:

            if attempt >= retries:

                raise

            delay = min(2 ** attempt, 30) * (0.5 + random.random())

            await asyncio.sleep(delay)

min(2**attempt, 30) 是上限，避免第 10 次重试睡 17 分钟；乘一个 0.5–1.5 的随机因子是「抖动」，防止所有协程在同一时刻集体重试，形成二次冲击。这是 AWS 那篇讲指数退避与抖动的文章里最值得抄的一条。

另外 429 响应通常带 Retry-After 头，优先级应该高于自己算的退避时间。

结果收集不要用 gather

gather 默认 return_exceptions=False，任何一个协程抛异常，其余协程的结果就全丢了。用 as_completed 边完成边处理，失败也能单独记录：

for coro in asyncio.as_completed([fetch_with_retry(u) for u in urls]):

    try:

        html = await coro

    except Exception as e:

        failed.append(str(e))

如果一定要用 gather，至少传 return_exceptions=True，然后逐个判断返回值是不是 Exception。

实测数据

同一个接口列表，1000 个 URL，单机 2 核：

| 并发 | 耗时 | 失败率 |

|---|---|---|

| 无限制 | 直接 fd 耗尽 | — |

| 20 | 48s | 0.2% |

| 50 | 21s | 0.3% |

| 100 | 14s | 4.1%（触发对端限流） |

并发不是越高越好，找到拐点比一味调大更有意义。20–50 这个区间对大多数第三方接口都是安全的，具体还要看对端的限制策略和 SLA。