Python 多进程 Queue 传输结果的关闭方式

Python 的 multiprocessing.Queue 适合在多个进程之间传递结果，但它不是一个“调用 close() 就立刻结束”的普通列表。对象会先被序列化，生产进程还可能通过后台 feeder thread 把缓冲区里的数据刷入管道。正确的收尾顺序通常是：主进程先持续 get()，确认生产端都发出结束标记后再 join() 子进程，最后关闭主进程自己的队列句柄。

官方地址：https://docs.python.org/3/library/multiprocessing.html

下面用一个“多个 worker 返回结果”的小场景说明这套顺序。重点不是把所有进程 API 罗列一遍，而是解决结果已经放进 Queue 后，程序退出时为什么还会卡住的问题。

Queue 关闭前要先理解 feeder thread

multiprocessing.Queue.put() 并不等于调用进程已经把完整对象同步写入管道。Queue 会对对象做序列化，并在需要时启动后台 feeder thread，将进程内的缓冲内容继续送往管道。消费进程通过 get() 读取的是已经进入队列传输通道的数据。

因此，close() 的含义是释放当前进程持有的 Queue 内部资源，并且这个进程不能再对它调用 put() 或 get()。它不是向其它进程广播“所有数据都已被读取”的信号，也不替代进程间的结束协议。

先约定结果和结束标记

多个 worker 时，主进程不能只依赖 empty() 判断任务是否结束。队列可能正在接收数据，empty() 也不适合作为可靠的跨进程完成条件。更稳妥的方式是让每个 worker 在结果之后放入一个 None sentinel，主进程收到与 worker 数量相同的 sentinel 后，才认为生产端全部结束。

这个约定把“业务结果”和“生产者结束”区分开：普通字典是业务数据，None 只负责结束计数。若业务本身允许返回 None，就改用一个明确的字典标记或专用对象。

一套可复用的安全收尾代码

下面的例子使用 spawn 上下文，显式展示跨平台进程入口。worker 完成最后一次 put() 后关闭自己持有的队列句柄，并等待自己的 feeder thread 刷完；主进程则一边消费，一边等待所有 sentinel。

import multiprocessing as mp


def worker(result_queue, worker_id):
    try:
        # 结果对象会被序列化后交给 Queue 的传输缓冲区。
        result_queue.put({"worker": worker_id, "value": worker_id * 10})
        # sentinel 只表示这个生产者不再提交新的结果。
        result_queue.put(None)
    finally:
        # 当前 worker 不再使用自己的 Queue 句柄，先释放内部资源。
        result_queue.close()
        # 等待本进程 feeder thread 把已入队数据刷入管道。
        result_queue.join_thread()


if __name__ == "__main__":
    # spawn 要求进程创建代码放在保护块中。
    context = mp.get_context("spawn")
    result_queue = context.Queue()
    processes = [
        context.Process(target=worker, args=(result_queue, worker_id))
        for worker_id in range(3)
    ]

    for process in processes:
        # 先启动生产者，再由主进程持续取结果。
        process.start()

    results = []
    finished_workers = 0
    while finished_workers  join_thread 顺序清理。
    result_queue.close()
    result_queue.join_thread()
    print(sorted(results, key=lambda item: item["worker"]))

这段代码的关键点是主进程没有在 process.join() 前停止读取 Queue。对于单个小结果，错误顺序可能暂时看不出问题；当结果较大或数量较多时，子进程可能还在等待 feeder thread 刷空缓冲区，主进程却先等待子进程结束，于是形成互相等待。

close、join_thread 和 process.join 分别负责什么

可以把三个方法分成三种对象来理解：

生产端和消费端各自持有 Queue 的进程内句柄，所以 worker 的 close() 不会让主进程的 get() 失效。主进程必须等自己的消费工作完成后再关闭自己的句柄。若只有一个生产者，也可以用一个 sentinel；如果任务数量本身固定，也可以按固定结果数读取，但仍要保证所有生产者的进程回收顺序不会堵住未消费的数据。

最容易踩中的两种关闭错误

错误一：先 join 子进程，再读取 Queue

# 可能造成阻塞：子进程的缓冲数据还没有被消费。
process.join()
item = result_queue.get()

当子进程向 Queue 放入的数据较大时，子进程退出前需要 feeder thread 把缓冲内容送入管道。主进程先 join()，就没有机会继续 get() 腾出传输空间。修复方式是把 get() 放在前面，或者使用独立的协调协议，确保消费者不会停读。

错误二：close 后继续 put 或 get

# close 表示当前进程不再使用这个 Queue。
result_queue.close()
# 这里继续 put/get 属于错误的生命周期使用。
result_queue.put({"late": True})

如果业务还需要发送数据，说明关闭时机太早，应把 close() 放到最后一次 put() 之后。关闭后是否还有其它进程在工作，也不改变当前进程不能继续使用该句柄的规则。

cancel_join_thread 什么时候可以用

cancel_join_thread() 会阻止进程退出时自动等待 Queue 的后台线程。它适合明确接受“进程可以马上退出，未刷完的数据可能丢失”的场景，例如放弃一批无关紧要的临时统计。它不是加速版的 join_thread()，也不是常规生产代码的默认清理方式。

# 只有明确允许丢弃尚未刷出的临时数据时才这样做。
result_queue.cancel_join_thread()
result_queue.close()

正式结果、任务状态和需要审计的消息都不应依赖这个方法保证退出速度。若目标是可靠传输，应让消费者读完、生产端关闭并等待 feeder thread，再回收进程。

排查 Queue 关闭问题的顺序

1. 确认每个 worker 是否都在最后一次结果后发送了结束标记。
2. 确认主进程是否持续 get()，而不是先对生产者调用 join()。
3. 确认 worker 的 close() 和 join_thread() 位于最后一次 put() 之后。
4. 确认主进程收到所有 sentinel 后才回收进程，并最后清理自己的 Queue。
5. 只有在明确允许数据丢失时，才考虑 cancel_join_thread()。

记住一句话：Queue 的“关闭”是某个进程释放自己的使用权，进程的“join”是等待子进程结束，二者都不能代替消费协议。把结束标记、消费循环和清理顺序写清楚，多进程传输结果就不容易在退出阶段变成一个隐蔽的死锁。

相关问题

Queue.close() 会通知其它进程停止吗？

不会。它只关闭调用者当前持有的 Queue 资源；跨进程停止需要 sentinel、事件或其它明确协议。

为什么 Queue.put() 成功了，进程还是退出很慢？

因为 put 后的数据可能仍在 feeder thread 的缓冲区中，进程退出时需要等待刷入管道。主进程应继续消费，生产端再按 close 和 join_thread 收尾。