asyncio.Semaphore 限住并发后，为什么任务仍然占满内存

给每次请求套上 asyncio.Semaphore(10)，同时运行的请求确实少了，内存却仍然随着输入条数增长。问题可能没有发生在许可数量上，而是代码先为全部输入创建了 Task，随后才让这些 Task 在信号量前排队。

信号量限制进入受保护区域的执行者数量，不限制任务对象、闭包参数和输入载荷的数量。十个任务在跑，十万个任务等许可，依然是十万个任务的生命周期。

下面固定 Python 3.12，用一组纯本地异步计算对照两种调度方式。实验只比较确切创建的任务数量与活动工作数，不把计数冒充内存或吞吐实测。然后给出一个滚动提交实现，把在途 Task 数也限制在上限内。

先创建全部任务，许可只管后半段

常见写法是对所有输入调用 create_task()，每个任务进入函数后再 async with semaphore。这样可以控制受保护区并发，但创建任务的循环没有背压。

若输入早已在一个大列表里，任务对象之外还有输入本身；若每项还捕获了响应、文件内容等对象，等待者又会延长它们的寿命。仅调整许可，未必能碰到真正占内存的那一层。

滚动提交的思路更直接：最多保留 N 个在途任务，等待至少一个完成，取走其结果，然后才从输入迭代器取下一项。下面把两种方式放在同一个完整脚本里。

import asyncio


class Gauge:
    def __init__(self):
        self.active = 0
        self.peak = 0

    async def work(self, value):
        self.active += 1
        self.peak = max(self.peak, self.active)
        try:
            await asyncio.sleep(0)
            if value < 0:
                raise ValueError("negative input")
            return value * 2
        finally:
            self.active -= 1


async def create_all(values, limit, gauge):
    semaphore = asyncio.Semaphore(limit)

    async def guarded(value):
        async with semaphore:
            return await gauge.work(value)

    tasks = [asyncio.create_task(guarded(value)) for value in values]
    created_before_wait = len(tasks)
    results = await asyncio.gather(*tasks)
    return results, created_before_wait


async def rolling_map(values, limit, work):
    if limit <= 0:
        raise ValueError("limit must be positive")
    source = iter(enumerate(values))
    pending = {}
    results = {}
    submitted = 0
    peak_tasks = 0
    exhausted = False

    def fill():
        nonlocal submitted, peak_tasks, exhausted
        while not exhausted and len(pending) < limit:
            try:
                index, value = next(source)
            except StopIteration:
                exhausted = True
                break
            task = asyncio.create_task(work(value))
            pending[task] = index
            submitted += 1
            peak_tasks = max(peak_tasks, len(pending))

    try:
        fill()
        while pending:
            done, _ = await asyncio.wait(
                pending, return_when=asyncio.FIRST_COMPLETED
            )
            for task in done:
                index = pending[task]
                results[index] = task.result()
                del pending[task]
            fill()
    except BaseException:
        for task in pending:
            task.cancel()
        await asyncio.gather(*pending, return_exceptions=True)
        raise
    return [results[index] for index in range(submitted)], peak_tasks


async def main():
    first = Gauge()
    values, created = await create_all(range(100), 3, first)
    assert values == [value * 2 for value in range(100)]
    assert created == 100 and first.peak <= 3
    print("create all: 100 tasks created before waiting, active <= 3")

    second = Gauge()
    values, peak = await rolling_map(range(100), 3, second.work)
    assert values == [value * 2 for value in range(100)]
    assert peak <= 3 and second.peak <= 3 and second.active == 0
    print("rolling: in-flight tasks <= 3, active <= 3")

    failed = Gauge()
    try:
        await rolling_map([1, -1, 2, 3, 4], 3, failed.work)
    except ValueError:
        assert failed.active == 0
        print("failure: submitted tasks were cancelled and awaited")
    else:
        raise AssertionError("negative input was accepted")


if __name__ == "__main__":
    asyncio.run(main())

保存为 rolling_tasks.py，执行 python3 rolling_tasks.py。第一种方式在第一次等待之前已经创建 100 个 Task；第二种的在途数量最多为 3。两者活动工作数都不超过 3，这恰好说明活动并发与任务数量是不同指标。

FIRST_COMPLETED 返回的可能不止一个任务

事件循环恢复主任务时，可能已经有多项完成，因此 done 是集合，不是一份单独结果。代码处理这一批完成项，再补足容量，不能只拿出一个任务就忽略其余完成者。

结果根据输入编号存放，最终按编号返回。完成顺序不稳定，并不应该让输出与输入失去对应关系。若调用方本来就接受完成顺序，可以改成逐项产出，进一步减少结果积累；若必须严格输入顺序流式输出，则还要为前面的慢任务保留后续已完成结果。

后一种保序缓冲仍可能增长。限制任务数，并不会让所有保序需求免费变成恒定内存。最早一项迟迟不结束，后面即使不断有结果，也不能在不改变合同的情况下越过它。

示例返回列表，因此明确保留了全部结果。它解决的是任务对象的数量，不宣称整批内存完全有界。输入与输出都很大时，应继续设计迭代结果、受限存储或分批提交。

首错停止以后，还要把已提交任务等完

task.result() 可能抛出业务错误或编程错误，输入迭代器也可能在 next() 时失败。退出路径会取消仍登记在 pending 里的任务，再 gather 它们，最后重新抛出原异常。

这里捕获 BaseException 是为了清理以后继续传播，不是吞掉取消。若只调用 cancel 而不等待，函数已经向上报告失败，工作者还可能在 finally 里释放资源；调用方很难判断它们究竟结束了没有。

完成集合中也可能有另一项已经失败。清理阶段的 gather 会收集剩余结果，避免把已完成异常遗弃成未检索任务错误。对业务需要保留多个失败原因的场景，可以增加明确的错误结果账本，但不应随手让第二个错误覆盖最初触发终止的那个。

取消不会撤销外部写入，线程或扩展中的阻塞工作也未必随协程立即停止。示例的计算通过 await 提供协作点，才适合用这条取消链验证退出。把同样代码外面包一个阻塞函数，并不会自动获得同样的响应性。

输入迭代本身也可能成为阻塞点

滚动实现从同步迭代器逐项取值，适合range、普通容器或每次next很快的本地生成器。如果next内部读取慢文件、等待网络或执行大量CPU计算，fill仍会堵住事件循环。限制Task数改变不了同步输入函数的执行方式，这时需要异步输入协议或明确的线程桥接，而不是再给next外面套一层信号量。

输入生成器若自己拥有打开的文件，谁关闭它也要事先决定。当前函数只消费调用方传入的可迭代对象，不声明接管任意迭代器的close责任。调用方可以在外层with里建立输入资源；如果将函数改成自己打开文件，就应在函数退出时负责关闭，不能把取消下游任务当成关闭上游资源的替代动作。

同一类边界也作用于结果消费。现在每个结果都留在字典里，改成边算边写时，写入失败应进入同一条取消和等待路径。若在成功结果回调中私自创建后台任务，又不把它登记进在途集合，滚动窗口就管不到那些任务了。容量合同必须覆盖真正提交工作的入口，而不是只覆盖其中一层漂亮的API。

单项超时可以放进work自己的作用域，让到期错误按同样路径触发整批退出，或转换成逐项失败。但超时发生以后是否允许其他项目继续，要保持一贯政策。一次超时并不总说明远端没有处理，不应在窗口补位时自动生成无限重试链。

信号量仍然有它合适的位置

多个独立流程共享同一类稀缺资源时，信号量很有用。比如不同批次共用最多十个远程连接，可以共享一个许可池；单批滚动任务数各自限制，并不能天然表达跨批次总量。

也可以同时使用两层限制：任务提交层控制对象数量，资源访问层控制实际占用。但要明确获取顺序，避免任务拿着一种许可又无限等另一种许可，造成资源互相占住。

许可数还不是请求速率。十个并发任务可能每秒完成很多次，也可能长时间一项都不结束。若下游合同是每秒最多多少请求，需要单独的速率限制，而不是继续降低 Semaphore 的值直到“感觉差不多”。

生产程序还应限制单项尺寸、失败后的重试数量和总处理预算。一次重试若绕过原来的提交入口，仍可能让在途数量失控；重试任务也必须算进同一份容量账目。

选择结构时，先定位自己想约束的是等待任务、活动工作、资源租约还是单位时间请求数。它们可以同时存在，却不是同一件事。用活动并发指标证明任务数量受控，正是这类内存问题最容易走偏的一步。