Python asyncio.Semaphore 怎么限制并发任务数量

批量请求、文件处理或异步任务一多，最容易出现的误会是：已经用 asyncio.gather() 创建了任务，就等于已经限制了并发。实际上，任务可以一次性创建很多个；真正需要限制的是同一时刻有多少个协程进入那段受限工作。asyncio.Semaphore 用一个内部计数器完成这件事：拿到许可后计数减少，许可用完时后续协程等待，退出受限区时再归还许可。

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

如果目标是“最多同时运行 N 个协程的某一段工作”，就把 asyncio.Semaphore(N) 放在共享任务函数外部，用 async with semaphore 包住受限区；需要手动控制时，必须把 release() 放进 finally。

先分清：限制任务数量，还是限制受限区

我在批量处理场景里更愿意把“任务总数”和“并发上限”拆成两个问题。任务总数描述要处理多少项，并发上限描述同一时刻允许多少项占用下游资源。比如有 1000 个 URL 待处理，可以创建 1000 个协程任务，但让每个任务在访问外部服务前先竞争一个容量为 20 的信号量。

信号量的初始值就是内部计数器的初始值。acquire() 成功时计数减一；当计数已经为零，新的 acquire() 会等待；release() 会把计数加一，并可能唤醒等待中的任务。初始值不能小于零，默认值是 1。它是 asyncio 原语，不是用来同步操作系统线程的锁。

最小写法：用 async with 自动归还许可

最稳定的入口是把信号量作为多个任务共享的对象，然后只包住真正需要限流的代码。下面的例子仍然一次创建多个任务，但每个任务只有进入 async with limit 后才占用一个并发名额。

import asyncio


async def fetch_one(item: int, limit: asyncio.Semaphore) -> str:
    # 只有 async with 内的工作受到并发上限约束。
    async with limit:
        # 这里替换成真实的异步 I/O；sleep 只是模拟等待。
        await asyncio.sleep(0.05)
        return f"完成 {item}"


async def main() -> None:
    # 允许同时进入受限区的任务最多为 3 个。
    limit = asyncio.Semaphore(3)
    tasks = [
        asyncio.create_task(fetch_one(item, limit))
        for item in range(10)
    ]

    # gather 等待所有任务；Semaphore 负责限制受限区，而不是限制任务创建。
    results = await asyncio.gather(*tasks)
    print(results)


asyncio.run(main())

这里的关键不是 range(10)，而是所有任务共享同一个 limit。如果每次调用 fetch_one() 都重新创建一个信号量，任务之间就各自拥有计数器，并不能形成全局上限。

手动 acquire 时，把 release 放进 finally

async with 已经覆盖了大多数情况，但有时需要在取得许可前后加入额外判断，或者必须把“等待许可”和“执行工作”分开记录。这时可以手动调用 acquire()，但配对关系不能靠记忆维护。

import asyncio


async def fetch_with_cleanup(item: int, limit: asyncio.Semaphore) -> str:
    # acquire 可能等待，因此许可取得前不应把任务算作正在处理。
    await limit.acquire()
    try:
        # 业务异常、CancelledError 或正常返回都会经过 finally。
        await asyncio.sleep(0.05)
        if item == 4:
            raise RuntimeError("模拟单项失败")
        return f"完成 {item}"
    finally:
        # 无论工作如何结束，都把这一次成功取得的许可归还。
        limit.release()

顺序很重要：只有 acquire() 成功返回后，才进入需要归还的 try 区域。若在等待许可阶段被取消，并没有拿到许可，也就不应该额外调用 release()。

超时应该包住 acquire，而不是假设它自带 timeout

asyncio 的同步原语方法本身不接收 timeout 参数。如果等待许可也必须有上限，可以用 asyncio.wait_for() 包住 acquire()。这能把“资源迟迟不可用”和“拿到资源后业务执行超时”区分开。

import asyncio


async def try_fetch(item: int, limit: asyncio.Semaphore) -> str | None:
    try:
        # 这里只给等待许可设置 0.2 秒上限。
        await asyncio.wait_for(limit.acquire(), timeout=0.2)
    except asyncio.TimeoutError:
        # acquire 超时意味着没有拿到许可，因此这里不能 release。
        return None

    try:
        # 许可已经拿到，后续工作必须由 finally 负责归还。
        await asyncio.sleep(0.05)
        return f"完成 {item}"
    finally:
        # 只释放本协程实际持有的许可。
        limit.release()

检查当前是否不能立即取得许可，可以读取 limit.locked()。它返回 True 时表示当前不能立即获取，但这不是一个预留名额的操作；检查结果和真正的 acquire() 之间仍可能有其他任务改变状态，所以它适合记录或调试，不适合代替获取许可。

Semaphore、BoundedSemaphore 和过度 release

普通 Semaphore 允许调用次数多于成功的 acquire() 次数，内部计数器可能因此超过初始值。这个特性有时便于表达“可增加的容量”，但如果代码把许可释放错了，问题不会立刻暴露成异常。

asyncio.BoundedSemaphore 会记住初始值；当 release() 让计数超过这个初始值时，它会抛出 ValueError。如果你的许可数量应该严格守恒，使用它能更早暴露配对错误。不过它依然需要正确的 acquire() 与 release() 生命周期，不能把异常交给信号量替你修复。

最后用一张清单检查并发上限

• 信号量是否在任务之间共享，而不是在每次函数调用中重新创建？
• 是否只把真正需要限流的 I/O 或资源占用放进 async with？
• 手动调用 acquire() 后，是否只在成功取得许可后进入 try/finally？
• 异常、取消和正常返回是否都会执行 release()？
• 等待许可超时后，是否避免释放一个实际上没有取得的许可？
• 是否把“同时运行的协程数”与线程池、进程池或下游服务的其他容量限制混在了一起？

把这几个问题逐项回答清楚，通常就能判断 asyncio.Semaphore 是否放在了正确的边界上。它控制的是协程进入共享受限区的许可数量；任务如何创建、异常如何传播和业务如何重试，仍然需要由调用方自己设计。

相关问题

Semaphore 的默认并发数是多少？

构造函数默认值是 1，因此默认表现接近一次只允许一个任务进入受限区；生产代码应根据下游容量明确传入数值。

为什么 release 后并发数看起来超过了设定值？

常见原因是重复调用了 release()。普通 Semaphore 允许过度释放；需要尽早发现配对错误时可考虑 BoundedSemaphore。

asyncio.Semaphore 能限制线程池任务吗？

本文范围内不能把它当作操作系统线程同步工具。它面向 asyncio 任务；线程同步应选择线程模型对应的同步原语，并单独设计跨模型边界。