Python 任务队列怎么选：queue.Queue、asyncio.Queue 与 multiprocessing.Queue

同一个接口的后台任务一多，很多项目都会先塞进一个 Queue：爬取结果要慢慢落库，上传后的转码不能堵住请求，批处理又想开几路工人。问题是，线程函数里能正常工作的 queue.Queue，直接拿到协程里往往会把事件循环卡住；而把小字典频繁送进 multiprocessing.Queue，又会多出序列化和进程收尾的成本。先看任务究竟跑在线程、同一事件循环，还是另一个进程里，队列类型就不会选错。

先确认任务在哪儿跑，别从 API 名字开始挑

选型可以先问一句很朴素的话：消费者拿到任务后，接下来那段代码是什么形态？如果是阻塞式 SDK、文件读写或同步数据库驱动，并且你已用多个线程分担它，queue.Queue 的锁和条件等待正好匹配这个模型。假如消费者的主要动作是 await client.get(...)、await writer.drain() 这类异步 I/O，它们共享一个事件循环，应该用 asyncio.Queue。只有工作真的需要在独立进程里跑，例如图像编码、科学计算或隔离第三方库，才让 multiprocessing.Queue 跨进程搬运数据。

这张表的核心不是“谁更快”。同一事件循环里为了躲开 await 去开进程，通常只会增加数据搬运；CPU 已经跑满却硬塞线程，也不会让解释器把同一段 Python 计算真正并行起来。任务的阻塞点和隔离边界，才是第一层判断。

协程任务先用 asyncio.Queue 做出可观察的背压

Web 服务或异步采集程序最常见的误区，是把无界队列当作“请求已经接收成功”。消费者比生产者慢时，任务其实只是从网络缓冲区搬到了 Python 内存里。给 asyncio.Queue 一个合适的 maxsize，再给生产动作设置等待上限，队列满时就会把压力传回调用路径。这里的数值没有通用答案，应该按单任务体积、消费者数量和可接受的排队时间压测后确定。

import asyncio


async def save_job(job: dict) -> None:
    await asyncio.sleep(0.05)  # 替换为真实异步 I/O


async def worker(name: str, jobs: asyncio.Queue[dict]) -> None:
    while True:
        job = await jobs.get()
        try:
            await save_job(job)
            print(name, "done", job["id"])
        finally:
            jobs.task_done()


async def submit(jobs: asyncio.Queue[dict], job: dict) -> None:
    await asyncio.wait_for(jobs.put(job), timeout=0.5)


async def main() -> None:
    jobs: asyncio.Queue[dict] = asyncio.Queue(maxsize=20)
    workers = [asyncio.create_task(worker(f"w{i}", jobs)) for i in range(3)]

    for job_id in range(10):
        await submit(jobs, {"id": job_id})

    await jobs.join()
    for task in workers:
        task.cancel()
    await asyncio.gather(*workers, return_exceptions=True)


asyncio.run(main())

先跑这段示例时，重点看两个结果：jobs.join() 只有在每个 get() 对应一次 task_done() 后才会返回；而 asyncio.wait_for() 才是队列操作的超时入口，asyncio.Queue 自己的 put() 没有 timeout 参数。少一次 task_done()，等待就会一直挂着；多一次则会报错。这比盯着 qsize() 更能说明任务是否真的收完。

线程队列适合阻塞库，但也要把收尾写进 worker

同步 HTTP 客户端、旧版 SDK 或必须调用阻塞文件接口时，线程仍是很实用的选择。标准库 queue.Queue 为多生产者、多消费者提供同步语义，maxsize 大于零时，队列放满后 put() 会等待空位。这里先别急着用 empty() 判断“没有任务了”：官方文档明确说明，观察到空或非空并不保证下一次 get() 或 put() 一定不会等待。

from queue import Queue
from threading import Thread


jobs: Queue[str] = Queue(maxsize=50)


def worker() -> None:
    while True:
        path = jobs.get()
        try:
            write_thumbnail(path)  # 同步阻塞库
        finally:
            jobs.task_done()


Thread(target=worker, daemon=True).start()
jobs.put("/tmp/a.jpg")
jobs.join()

这个模式的关键在 finally。即使 write_thumbnail() 抛出异常，已取出的任务也要完成一次状态收尾；否则主线程会在 join() 前停住，而日志里只剩下一段与真正原因隔得很远的等待。真实服务还应把异常记录、失败重试和停止信号设计清楚，不能让守护线程无限吞掉错误。

跨进程时，先估算序列化成本，再决定是否搬运对象

multiprocessing.Queue 适合把工作送到另一个进程，但它不是线程队列的无缝替换。Python 会把放入其中的对象序列化，接收端得到的是重新构造的对象；大量大字典、图像字节或深层对象会让这一步变成可见成本。普通 multiprocessing.Queue 也没有 task_done() 和 join()，确实需要“所有任务都处理完”的语义时，应当直接选择 multiprocessing.JoinableQueue。

进程模式里还有一个容易被忽略的收尾点：子进程写进队列的数据会由后台线程送入管道。如果父进程过早等待子进程退出，而消费者还没有把缓冲数据拿走，就可能互相等住。先保证消费端能持续取走数据，再安排进程收尾；不要依赖 qsize()、empty() 或 full() 做进程状态机的分支条件，尤其在 macOS 上 qsize() 还可能不可用。

把选择落到一个小决策表里

如果当前项目的真实问题只是“消费端太慢”，先用有界队列、处理耗时和失败数把现象看清，再考虑换并发模型。队列类型不是性能补丁；它只是把任务所有权、等待关系和收尾责任放在合适的边界里。

相关问题

asyncio.Queue 能从另一个线程直接 put 吗？

不建议。它不是线程安全容器，跨线程投递应回到事件循环提供的线程安全调度方式，或在两个边界之间放置明确的桥接层。

Queue 的 maxsize 应该填多大？

从内存预算和可接受排队时间反推。先记录单任务大小、消费者处理耗时和队列等待时间，再用压测调整；无限大只会把背压延后。

为什么 task_done 和 join 总是卡住？

先逐个核对：每次成功 get() 是否恰好有一次 task_done()，异常路径是否走进 finally，以及有没有在任务还没取完时提前停止消费者。

什么时候不该用进程队列？

任务主要是网络等待、数据很大且频繁传递，或必须共享复杂的进程内对象时，进程边界通常会放大成本。先试异步 I/O 或线程，再看 CPU 与隔离需求是否足够明确。

收尾时只核对三件事

先找任务运行的边界，再给队列设置能暴露压力的容量，最后把每个 get() 的完成动作写进异常安全的收尾路径。线程用 queue.Queue，同一事件循环用 asyncio.Queue，跨进程才用 multiprocessing.Queue。这三个判断落稳了，后面的并发数量、重试和监控指标才有可靠的基础。

想复查细节时，可以阅读 Python 官方关于 线程队列、异步队列 与 进程队列 的说明；真正上生产前，仍应以目标机器、Python 版本和任务负载的压测结果为准。