LLM 请求一多，先别把所有请求都包装成协程再加信号量。信号量可以限制同时执行多少次调用，却不自动限制有多少任务已经在内存里等待。用有界队列控制接纳，再用固定数量的工作协程消费，才能把等待容量和执行容量分别说清楚。

慢服务前面，最容易藏一条无限队列

假设一个应用每收到请求，就立即创建一个等待模型调用的任务。下游变慢后，同时执行的调用虽然不多，等待任务却继续增长。它们可能持有提示词、会话状态、回调和超时上下文；没有发出网络请求，不等于没有资源成本。

这也是只看“活跃连接数”容易漏掉的问题。服务没有立刻崩溃，等待时间却越来越长，用户取消后工作还留在后台。继续扩大协程数通常不会让有限的下游吞吐凭空增加，反而会积累更多已经来不及完成的请求。

背压要把“暂时不能接收更多工作”传回上游。它可以表现为生产者等待、有限时间内接纳，或明确拒绝。关键是等待也有边界，而不是把压力从一个列表换到另一个看不见的任务集合。

可以先做一个容量算例：假设持续到达12个同等工作量的请求每秒，而处理能力稳定为每秒8个；忽略启动边界，净积压约每秒4个，10秒约40个。这个差额不会因为队列改成1000格而消失。大队列只是延后满载，同时允许更多工作等待；有界队列则迫使入口尽早暴露能力不足。

这些速率是人为假设，不是模型服务测试。真实请求长度不同、速率波动，还可能按调用者分配配额。容量判断应同时看持续负载和突发，不把一次短暂峰值当成必须永久扩容的证据，也不把均值尚可当成尾部等待已经可接受。

三个数不要混成一个并发参数

第一是队列容量：有多少已接纳、尚未被工作协程取走的任务。第二是执行并发：同时有多少任务正在调用下游。第三是服务自己的限额，例如请求速率或 token 预算。把并发设为2，不代表每秒只发2次；短请求可以在一秒内周转很多次。

Python 的 asyncio.Queue 文档说明，正的 maxsize 会让满队列上的 put 等待；默认0则不提供这个容量限制。本例只用两个固定工作协程，因此下游模拟调用也最多同时2个。它不靠为每个输入都创建一个新任务来等待。

但队列上限2不等于整个进程最多保存2个请求。还有正在执行的2个、生产者手里的一个，以及入口框架可能缓存的工作。真实 HTTP 服务若让每条连接无限等待入队，仍可能积累大量入口协程。因此接纳等待、连接数、请求体尺寸和整体超时都要一起设计。

等位时间也应占用总预算

一次用户请求可能经历入队等待、检索、模型调用与后处理。如果每一层都重新给出完整的30秒，最终等待就可能远超用户愿意接受的时间。更清楚的做法，是从入口确定总截止时间，让后面的步骤使用剩余预算。

队列里已经过期的任务不应继续消耗昂贵调用。工作协程取到任务后，需要检查它是否仍有效，再决定执行或记为过期。下面程序为清楚展示容量和记账，只对模拟调用设置等待上限；它没有实现端到端截止时间，不能直接当成完整网关。

如果选择拒绝，返回语义也要可解释：是暂时满载、调用者自己的配额用尽，还是请求本身无效。不要把所有情况都包装成“模型出错”，否则上游可能对无效请求反复重试。对有副作用的工具，超时后的安全重放还要遵循幂等约定。

取走不等于完成，失败也要记账

Queue.get 把任务从等待队列取出，队列长度会下降，但这时任务可能刚开始执行。把 qsize 当成总在途量，会在繁忙时低估负载。排队数与执行数最好分别观察。

每个成功取出的任务，最终都需要恰好一次 task_done，join 才能知道接纳的工作已经处理完。这里的“完成”是队列计数，不是业务成功；失败结果也需要保存并完成记账，否则 join 可能一直等着一个再也不会减少的数字。

把 task_done 放在与 get 对应的 finally 里，有助于覆盖正常返回、受控失败和取消清理。但如果任务在 get 之前就被取消，不能凭空调用 task_done。下例也为停止标记做了对应记账，避免退出分支漏掉计数。

本地可运行的有界工作池

import asyncio

async def check_boundaries():
    q = asyncio.Queue(maxsize=1)
    q.put_nowait("first")
    blocked = asyncio.create_task(q.put("second"))
    await asyncio.sleep(0)
    assert not blocked.done()
    assert q.get_nowait() == "first"
    q.task_done()
    await blocked
    assert q.get_nowait() == "second"
    q.task_done()
    await q.join()

    cleanup = []
    async def never_finishes():
        try:
            await asyncio.Event().wait()
        finally:
            cleanup.append(True)
    try:
        await asyncio.wait_for(never_finishes(), timeout=0.01)
    except asyncio.TimeoutError:
        pass
    else:
        raise AssertionError("Expected timeout")
    assert cleanup == [True]

async def main():
    await check_boundaries()
    q = asyncio.Queue(maxsize=2)
    active = peak = max_queued = 0
    outcomes = []

    async def mock_call(item):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)
            if item == 3:
                raise ValueError("Simulated invalid request")
            return f"result-{item}"
        finally:
            active -= 1

    async def worker():
        while True:
            item = await q.get()
            try:
                if item is None:
                    return
                try:
                    await asyncio.wait_for(mock_call(item), timeout=1)
                except (asyncio.TimeoutError, ValueError) as error:
                    outcomes.append((item, type(error).__name__))
                else:
                    outcomes.append((item, "ok"))
            finally:
                q.task_done()

    workers = [asyncio.create_task(worker()) for _ in range(2)]
    for item in range(6):
        await q.put(item)
        max_queued = max(max_queued, q.qsize())
    for _ in workers:
        await q.put(None)
    await q.join()
    await asyncio.gather(*workers)
    assert 1 <= peak <= 2 and max_queued <= 2
    assert active == 0 and len(outcomes) == 6
    assert sum(status == "ok" for _, status in outcomes) == 5
    print("peak active:", peak, "max queued:", max_queued)
    print(sorted(outcomes))

asyncio.run(main())

原创 Python 程序已执行。边界测试验证满队列确实阻塞生产者，取走后才放行；另一个本地测试验证 wait_for 超时会触发被取消协程的 finally。工作池处理6个模拟任务，其中1个按设计失败、5个成功，最大执行数和等待数均不超过2。

程序没有调用真实模型服务，没有测吞吐，也没有把模拟失败当成模型能力结果。outcomes 为了展示结果保存在列表中；无限运行的服务必须持续输出或限制结果存储，不能让结果列表变成新的无界队列。停止标记使用 None，所以业务任务不能同时把 None 当作有效输入。

取消本地协程，不代表远端已经停工

asyncio 的任务文档要求取消时正确清理并传播取消信号。对网络请求而言，本地停止等待与远端撤销计算仍是两件事。客户端是否关闭连接、服务是否继续生成、费用如何计量，都取决于接口与实现。

因此，观察到超时后，不应直接断言资源已经释放或请求没有执行。对于流式响应，消费者退出后也要按所用客户端的约定关闭流，检查后台任务是否退出。对写入型工具，则需要查询或重放原操作结果，不能因等待结束就创建一个新意图。

本例让局部 ValueError 变成一条失败结果，继续消费后续任务；没有吞掉整个工作协程的取消。生产代码还要安排工作协程异常退出后的监督、待处理任务的去向和服务关闭流程，避免只剩下一个永远无法排空的队列。

长短请求混排时，数量上限还不够

两条很短的提示词与两条接近上下文上限的请求，对内存、token和完成时间的压力不同。纯请求数限制容易把这种差异藏起来。可以在接纳前限制输入大小，再按实际业务评估是否需要分组、优先级或加权预算。

这些策略也有代价。只优先短请求可能让长请求长期等不到，简单先到先服务又可能让短请求被长任务挡住。不要直接把某个队列算法视为普遍最优；先观察请求分布与服务目标，再比较关键类别的等待与完成情况。

应用层队列还不同于模型引擎的连续批处理。前者决定哪些工作被接纳与何时发出调用，后者在推理引擎内部调度生成。调整客户端工作协程数量，不能替代引擎层的显存与批次管理。

上线前看四条曲线，而不只看总吞吐

分别记录等待队列长度、执行调用数、入队到开始的时间，以及完成、失败、过期或拒绝的数量。队列长期不下降、等待持续增长，即使成功请求的平均调用时长看起来正常，也说明入口与处理能力不匹配。

测试应包含突发输入、慢调用、失败、取消和正常停机。每个已接纳任务都应有最终分类；不能只统计成功完成的那部分，再宣称系统延迟很低。结合评测分母与记录口径，先让负载和结果解释得通，再决定扩容或调整上限。

一个可用的背压方案，不会承诺接住所有请求。它会让能完成的工作进入系统，让已经超出容量的工作得到明确结果，并把每一份等待放在可观察、有边界的位置。这样下游变慢时，问题才不会先在你的应用内存里悄悄扩大。

Read the English version

参考资料

• Python3.13 — asyncio queues
• Python3.13 — Semaphores
• Python3.13 — Task cancellation

本文于2026年10月14日晚补发，文章日期保留原计划北京时间19:00。代码只运行本地模拟任务，不是模型服务流量实测。