Python asyncio.Queue 如何实现可控生产消费：队列上限与优雅停机

批量抓取任务最容易出现的失控场景，不是消费者写错了，而是生产者瞬间把几万条 URL 塞进内存，消费者还在按固定速度处理。asyncio.Queue 的价值就在这里：用 maxsize 给任务流设上限，让 put 在队列满时主动等待；停机时再用哨兵值、task_done 和 join 把“已经取出”和“已经处理完”区分开。

一个可控的 asyncio 生产消费模型，要同时具备有界队列、明确的结束信号和完成确认；只调用 queue.get() 而不配对 task_done()，最终会让 join() 永远等不到结束。

先用 maxsize 把生产速度关进边界

队列默认是无界的，await queue.put(item) 很快就会返回。面对外部接口分页、文件扫描或数据库游标，这种写法会把“下游处理不过来”转成内存持续增长。把 maxsize 设置成一个能承受的小窗口后，队列满时 put 会等待，生产者自然降速。

import asyncio

async def produce(queue: asyncio.Queue[str], urls: list[str]) -> None:
    for url in urls:
        await queue.put(url)
    await queue.put(None)

async def consume(queue: asyncio.Queue[str]) -> None:
    while True:
        item = await queue.get()
        try:
            if item is None:
                return
            await fetch(item)
        finally:
            queue.task_done()

async def fetch(url: str) -> None:
    await asyncio.sleep(0.01)

async def main() -> None:
    queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=100)
    producer = asyncio.create_task(produce(queue, ["/a", "/b", "/c"]))
    workers = [asyncio.create_task(consume(queue)) for _ in range(2)]
    await producer
    await queue.join()
    await asyncio.gather(*workers)

asyncio.run(main())

这里的关键不是 100 这个数字，而是“队列里最多保留多少待处理项”必须成为一个明确决策。生产者如果比消费者快，put 会卡在队列边界；监控 queue.qsize() 时，持续接近 100 说明下游吞吐不足，而不是应该继续把上限调大。

消费者为什么必须在 finally 里调用 task_done

queue.join() 等待的是未完成任务计数归零。每次成功从队列取出一项，计数就需要通过一次 task_done() 抵消；如果处理函数抛异常后直接离开循环，这个计数会永久留下，主协程就会一直卡在 join。

把 task_done() 放进 finally 有两个好处：正常任务和失败任务都能结算，停止哨兵也不会留下未完成计数。业务是否重试、记录失败或转入死信列表，应该在 finally 之前完成，不要吞掉异常后假装任务成功。

用 sentinel 传递结束信号，再等待所有任务完成

生产者结束并不等于消费者可以立刻退出，因为队列里可能还有已排队的 URL。示例中生产者放入一个 None，两个消费者则需要两个 sentinel；每个消费者取到自己的哨兵后才结束。生产者任务结束后，主协程先等待 queue.join()，再等待消费者任务收尾，这个顺序能保住队列中已经接收的工作。

async def run(urls: list[str], worker_count: int = 2) -> None:
    queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=100)
    producer = asyncio.create_task(produce_many(queue, urls))
    workers = [asyncio.create_task(consume(queue)) for _ in range(worker_count)]

    await producer
    for _ in workers:
        await queue.put(None)
    await queue.join()
    await asyncio.gather(*workers)

async def produce_many(queue: asyncio.Queue[str | None], urls: list[str]) -> None:
    for url in urls:
        await queue.put(url)

如果把 sentinel 放在真实任务之前，消费者会提前退出，后面的任务就没人处理。若只放一个 sentinel，只有一个消费者能结束，另一个仍会等待下一项；所以 sentinel 数量要和消费者数量一致。

把队列方案放进真实任务时要观察什么

有界队列解决的是内存和下游速度不匹配，不会自动解决网络超时、重复消费或任务幂等。工程上至少记录四个信号：qsize() 的高水位、单项处理耗时、失败与重试次数，以及从投递到完成的总等待时间。

并发数也不能无限增加。消费者数量应该受远端服务限流、连接池和本机 CPU 约束；如果只是把 worker 数量从 2 调到 50，却没有超时和取消边界，队列只会更快地制造一批同时失败的请求。

常见问题

asyncio.Queue 的 maxsize 是严格上限吗？

对通过 put() 写入的待处理项，它提供有界等待语义；读取侧的处理速度和正在执行的任务不计入队列长度，所以系统总在途任务数仍可能高于 maxsize。

为什么不直接 cancel 所有消费者？

取消可以用于紧急终止，但正常停机更适合用 sentinel，让已入队任务先完成并经过 task_done()。遇到外部故障时再结合超时和取消做快速回收。

queue.join() 能保证任务业务成功吗？

不能。它只确认每次 get() 都有对应的 task_done()，不代表远端请求成功。业务结果仍需单独记录成功、失败和重试状态。

小结：先定边界，再定停机协议

asyncio.Queue 的最小可靠组合是 maxsize、put、task_done、join 和与消费者数量匹配的 sentinel。先用队列上限控制积压，再用完成计数确认收尾，最后才根据耗时和失败率调整 worker 数量，这样生产消费系统才有可解释的运行边界。