Python asyncio 怎样限制并发请求数

批量调用接口时，直接对每个参数执行 asyncio.create_task() 很容易让任务同时撞向同一个服务。解决办法不是把任务改成串行，而是让任务继续并发调度，再用 asyncio.Semaphore 限制同一时刻进入单次 IO 的数量。

例如信号量初始值设为 3，最多只有 3 个任务持有许可并执行请求，其余任务在等待队列中排队。任务总数可以是 100，真正的并发请求数仍然由许可数控制。

官方地址：https://docs.python.org/3/library/asyncio-sync.html#asyncio.Semaphore

把 Semaphore 放在“每个任务的一次 IO”外面，用 async with 自动释放许可；不要把整个批处理函数或无关的 CPU 计算一起包进去。

先区分任务总数与并发上限

create_task() 负责把协程安排成任务，它不负责限制任务访问外部资源的数量。Semaphore 内部维护一个计数器：调用 acquire() 会消耗一个许可，计数为零时等待；调用 release() 会归还许可并唤醒等待任务。

一个适合批量请求的最小结构如下。示例用 asyncio.sleep() 代替真实网络客户端，重点是并发控制位置；代码块中的日志只是示意输出，不是本机运行结果。

import asyncio

async def fetch_one(item, semaphore):
    # 只保护一次外部 IO，避免把排队前的准备工作也算进占用量。
    async with semaphore:
        print(f"开始请求: {item}")
        await asyncio.sleep(0.2)
        return f"完成: {item}"

async def main():
    items = range(1, 11)
    # 初始值为 3，表示同一时刻最多三项进入受保护区。
    semaphore = asyncio.Semaphore(3)
    tasks = [asyncio.create_task(fetch_one(item, semaphore)) for item in items]
    # 等待并收集所有任务，避免只创建任务却提前退出事件循环。
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

这里会一次创建 10 个任务，但 fetch_one() 中的受保护区最多同时有 3 个任务。若只把 semaphore 放在任务创建循环外，或者在循环里复用同一个结果变量，反而会把任务关系和资源边界混在一起。

把信号量放在单次 IO 的边界上

并发限制的粒度应该对应被保护的资源。对同一个 API 设置一个信号量，对另一个数据库连接池可以设置另一个信号量；不要为了“全局安全”把所有无关工作共用一个小许可池，否则轻量任务也会被慢请求拖住。

async def fetch_json(client, url, semaphore):
    # 进入上下文后才占用远端请求额度，解析结果不必继续占用许可。
    async with semaphore:
        response = await client.get(url)
        # 先读取当前响应，再离开上下文归还许可。
        return await response.json()

async def collect(client, urls):
    # 只限制这个远端服务，不影响其他独立资源的任务。
    remote_limit = asyncio.Semaphore(8)
    tasks = [
        asyncio.create_task(fetch_json(client, url, remote_limit))
        for url in urls
    ]
    return await asyncio.gather(*tasks)

如果客户端本身还有连接池上限，Semaphore 不是替代品，而是应用层的额外门槛。最终并发数会受到多个边界共同影响；通常先按对方服务的配额或自身连接池容量设值，再观察等待时间和错误率调整。

处理超时和取消时不要丢掉许可

官方文档推荐用 async with semaphore，它等价于显式 acquire() 加 try/finally 中的 release()。这点对超时和取消尤其重要：任务在 IO 中退出时，清理逻辑仍应归还许可；如果手写 acquire() 后忘记在所有异常路径释放，队列最终会永久卡住。

async def fetch_with_timeout(client, url, semaphore):
    # 超时只包住一次网络 IO，避免把排队时间误当成远端响应时间。
    async with semaphore:
        try:
            async with asyncio.timeout(5):
                return await client.get(url)
        except TimeoutError:
            # 记录当前 URL 后返回可识别的失败结果，不吞掉许可清理。
            return {"url": url, "error": "timeout"}

如果任务被取消，通常不要把 CancelledError 当普通业务错误吞掉；让上下文管理器完成退出，再由上层决定是否取消整组任务。Python 3.11 及以上还可以用 asyncio.TaskGroup 管理相关任务，让任务生命周期更集中。

用结果和等待状态判断并发值是否合适

并发数不是越大越好。可以把每项任务的结果区分为成功、超时、业务错误和取消，再结合等待队列是否长期堆积做调整。若服务端频繁返回限流错误，先降低许可数；若任务几乎没有等待但连接池已耗尽，应该检查下游连接配置，而不是继续增大 Semaphore。

还有三个边界容易被忽略：

• Semaphore 不是线程锁。 asyncio 同步原语用于事件循环中的任务，不用于操作系统线程之间的同步。
• 不要重复 release。 普通 Semaphore 允许释放次数超过获取次数，错误的手动释放会让实际并发超过设定值；需要严格检查时可考虑 BoundedSemaphore。
• 不要只创建任务不保留引用。 统一把任务放进列表并等待它们，异常才有明确的汇总位置。

相关问题

• Semaphore 的值是不是任务总数？ 不是，它只表示同时持有许可、进入受保护区的上限。
• 为什么并发数设为 8 仍然感觉请求变慢？ 任务可能在等待连接池、DNS 或对方服务响应；Semaphore 只控制你放置它的那段代码。
• 超时后要不要手动调用 release？ 使用 async with 时不要重复手动释放，离开上下文会执行清理。