Python 异步批量任务的并发控制与背压

用 asyncio 写批量任务，最常见的写法是 `asyncio.gather(*[task(i) for i in ids])`，本地跑一百条没问题，上线遇到五千条就开始超时、内存飙升、对端限流。问题不在 asyncio，而在于把"并发"当成了"无限并发"。

一、并发要有上限

`gather` 会一次性创建所有协程，事件循环会尽力推进它们，于是瞬间发出几千个请求。正确做法是用信号量把并发压到可控范围：

```python

sem = asyncio.Semaphore(16)

async def one(i):

    async with sem:

        return await fetch(i)

```

16 这个数字不是随便定的：它取决于对端的限流阈值、你的出口带宽、以及单请求的耗时。经验公式是 并发 ≈ 目标 QPS × 平均延迟（秒），再留 20% 余量。如果对端有明确的 QPS 限制，就用令牌桶而不是信号量，因为信号量控制的是"同时进行中"的数量，不控制速率。

二、背压：生产快于消费时必须能停下来

真正的背压出现在生产者-消费者模型里。用 `asyncio.Queue()` 时一定要给 `maxsize`，否则队列会无限增长，内存在消费不过来时直接爆掉。给 `maxsize` 之后，`await queue.put(x)` 在队列满时会自然地挂起生产者，这就是背压。

```python

q = asyncio.Queue(maxsize=200)

async def producer():

    for item in source():

        await q.put(item) # 队列满就等

    await q.put(None)

async def consumer():

    while True:

        item = await q.get()

        if item is None:

            break

        await handle(item)

        q.task_done()

```

三、超时、重试、取消要一起设计

没有超时的异步任务等于给自己埋定时炸弹。用 `asyncio.wait_for(task, timeout=10)` 包住单次请求；重试要区分可重试错误（超时、5xx、连接重置）和不可重试错误（4xx、签名错误），后者重试只会放大问题。重试还要加抖动退避，避免所有 worker 在同一时刻一起重来。

四、失败隔离与结果落盘

批量任务不能因为一条失败就整体崩掉。用 `asyncio.gather(..., return_exceptions=True)` 收集异常，把成功的写进结果文件、失败的写进失败清单，支持断点续跑。落盘时注意别每条都 fsync，攒批写能差出十倍性能。

五、别忘了事件循环之外的阻塞

最常见的性能杀手是混进了同步库：`requests`、`pymysql`、`open()` 大文件，都会阻塞整个事件循环，让并发形同虚设。要么换成异步版本（httpx、aiomysql、aiofiles），要么用 `asyncio.to_thread` 丢到线程池。

一句话总结：异步不是让代码变快，而是让"等待"变得便宜；一旦你把并发放开，就必须同时把上限、超时、背压、隔离四件事配齐，否则收益会被放大后的故障吃干净。