---
title: "异步任务越跑内存越大：用有界队列给生产者背压"
date: 2026-11-12
slug: asyncio-bounded-queue-backpressure
canonical: https://kernelranch.com/blog/2026/11/asyncio-bounded-queue-backpressure
markdown_url: https://kernelranch.com/blog/2026/11/asyncio-bounded-queue-backpressure.md
tags: ["python", "asyncio", "queues", "性能"]
---给十万条输入各创建一个任务，即使任务内部有信号量限制并发，等待中的任务仍占内存。使用固定数量的 worker 和有界队列，队列满时让生产者等待，能把尚未处理的数据控制在明确的范围内。

## 1. 先证明队列满时生产者会停下

保存为 queue_lab.py，运行 python queue_lab.py。需要 Python 3.11 的 TaskGroup。

```python
import asyncio

async def main():
    queue = asyncio.Queue(maxsize=2)
    await queue.put(1)
    await queue.put(2)
    attempted = asyncio.Event()

    async def put_third():
        attempted.set()
        await queue.put(3)

    task = asyncio.create_task(put_third())
    await attempted.wait()
    assert not task.done()
    assert await queue.get() == 1
    queue.task_done()
    await task
    assert queue.qsize() == 2
    print("full queue: producer waited; one read released it")

    jobs = asyncio.Queue(maxsize=4)
    total = 0
    processed = 0

    async def worker():
        nonlocal total, processed
        while True:
            item = await jobs.get()
            try:
                if item is None:
                    return
                await asyncio.sleep(0)
                total += item * item
                processed += 1
            finally:
                jobs.task_done()

    async with asyncio.TaskGroup() as group:
        for _ in range(3):
            group.create_task(worker())
        for item in range(100):
            await jobs.put(item)
        for _ in range(3):
            await jobs.put(None)
        await jobs.join()

    assert processed == 100 and total == 328350
    print("processed=100; sum=328350; workers=3; queue_capacity=4")

asyncio.run(main())
```

## 2. 容量预算怎么算

| 数据所在位置 | 本例上限 |
|---|---|
| 队列中等待的工作项 | 4 |
| worker 已取出且尚未完成的工作项 | 3 |
| 生产者等待 put 的当前工作项 | 1 |

因此，在输入按需生成、worker 不另行堆积任务的条件下，等待和处理中的工作项最多约 8 个。队列限制的是条数。每条内容大小差异很大时，还需要限制单项字节数。示例将结果累加到计数器，若把所有输出追加到列表，输出端仍会随总任务数增长。

第一组实测生产者停在第三次 put，消费者取走一项后它完成。第二组处理 100 项，总和为 328350。await sleep(0) 只用于模拟让出执行权，实际 worker 中应等待网络或磁盘操作。

## 3. 处理异常和停机

TaskGroup 中任一 worker 抛出异常，会取消同组剩余任务，并打断组内正在等待的生产者。退出时异常向调用方传播。本例采用整批失败语义，已完成的外部副作用需要业务层自行处理。

task_done 要与每次成功 get 配对，包括结束标记。它用于减少未完成计数，即使在 finally 中调用，也不能作为业务成功标记。需要重试的任务应单独记录失败结果，并设置重试上限。

正常停机时，生产者先停止接收新输入，再给每个 worker 投递一个结束标记，等待队列排空和 worker 结束。设置停机总预算后，超时取消剩余任务。需要进程崩溃后继续处理的工作，应存入持久队列并定义确认和重试协议。

## 小结

同时限制队列容量和 worker 数量，让生产者在积压时等待。验收时检查输入、处理中数据和输出三处是否都能保持容量上限。

## 数据来源与检索时间

检索时间 2026-11-12。

| 来源 | URL | 对应内容 |
|---|---|---|
| Python 3.11 官方文档 | https://docs.python.org/3.11/library/asyncio-queue.html | 有界 Queue、put、join 和 task_done |
| Python asyncio 任务文档 | https://docs.python.org/3.11/library/asyncio-task.html#task-groups | TaskGroup 的异常传播和取消 |

## 核验说明

- 主示例在本地 Python 3.11.9 下执行，检查正文中的断言与输出。
- 实验范围是文中构造的输入和本地运行条件，部署时需按实际数据规模与运行环境核验。