Asyncio 队列工作池

功能说明

当任务数量远大于并发上限时，把所有协程一次性启动会耗尽连接或内存。队列工作池把生产与消费解耦：主协程负责投递任务，若干工作协程从队列取任务执行，有界队列同时提供背压，队列满时投递方自动等待。

示例用 dataclass 描述任务，工作协程把处理结果追加到共享列表。结束时向队列投递与工作协程数量相同的 None 哨兵，每个工作协程读到哨兵后退出，从而避免遗漏或提前结束。

关键代码

worker 在 finally 中调用 task_done，无论正常处理、抛异常还是读到哨兵都会释放计数，保证 join 不会永久阻塞。run_jobs 先创建全部工作协程，再投递任务和哨兵，随后 join 等待队列清空，最后 gather 收集协程结束状态。队列容量设为工作协程数量的两倍，既能平滑突发，又能限制积压。

使用注意

哨兵数量必须等于工作协程数量，否则会有协程一直等待。队列元素类型包含 None，任务本身不要再用 None 表示。如果任务可能抛出异常，应在 worker 内部捕获并记录，否则 gather 会抛出第一个异常，其余任务仍然执行但结果可能不完整。共享列表在单事件循环内是安全的，多线程或多进程需要换用线程安全结构。投递阶段若队列满会挂起，必须确保此时消费者已启动。

验证方式

八个任务、三个工作协程时输出 processed 8，并逐行打印工作协程名与任务编号。断言检查编号集合恰好是零到七，说明没有重复也没有遗漏。可以增加任务数量并把休眠时间调大，观察有界队列让投递节奏自动放慢，而内存占用保持稳定。

代码示例

import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    identifier: int
    payload: str


async def worker(
    name: str,
    queue: "asyncio.Queue[Job | None]",
    results: list[tuple[str, int]],
) -> None:
    while True:
        job = await queue.get()
        try:
            if job is None:
                return
            await asyncio.sleep(0.01)
            results.append((name, job.identifier))
        finally:
            queue.task_done()


async def run_jobs(jobs: list[Job], workers: int = 3) -> list[tuple[str, int]]:
    if workers < 1:
        raise ValueError("workers must be positive")
    queue: asyncio.Queue[Job | None] = asyncio.Queue(maxsize=workers * 2)
    results: list[tuple[str, int]] = []
    tasks = [
        asyncio.create_task(worker(f"worker-{index}", queue, results))
        for index in range(workers)
    ]
    for job in jobs:
        await queue.put(job)
    for _ in tasks:
        await queue.put(None)
    await queue.join()
    await asyncio.gather(*tasks)
    return results


def main() -> None:
    jobs = [Job(index, f"payload-{index}") for index in range(8)]
    results = asyncio.run(run_jobs(jobs))
    print(f"processed {len(results)} jobs")
    for name, identifier in sorted(results, key=lambda row: row[1]):
        print(name, "->", identifier)
    assert sorted(identifier for _, identifier in results) == list(range(8))


if __name__ == "__main__":
    main()