Python multiprocessing.Pool 实现生产者消费者模式的正确方法

• 挂起的核心原因是哨兵发送逻辑完全不匹配消费者数量：原代码里每个生产者生产完只往队列塞1个None，2个生产者总共就发2个退出信号。当消费者开了3个，多出来的那个消费者永远等不到None触发退出，加上消费者里用block=False取数，拿不到就空转，既不退出也不触发池的任务完成，最终pool.join()会一直等这个没退出的任务，表现为程序挂起。
• 关于JoinableQueue的使用限制误区：并不是JoinableQueue不能和Pool配合，而是它底层的同步原语（锁、条件变量）不支持pickle序列化。如果你在主进程提前实例化队列，Unix下Pool的Worker是fork出来的，会自动继承队列句柄，完全可以正常用；只有运行时动态把队列对象作为异步任务参数传递时，因为传参需要序列化对象，才会抛出那个RuntimeError。

如果你的业务场景是基于Pool提交生产、消费的短任务，不需要长期常驻消费者，只要调整两个设计就能彻底解决挂起问题：

1. 不要让生产者自行发送退出哨兵：等所有生产者的异步任务全部执行完成后，由主进程统计实际启动的消费者总数，往队列里塞入对应数量的None，保证每个消费者都能拿到退出信号，不会漏发。
2. 优化消费者的取数逻辑，避免CPU空转：把非阻塞的立即取数改成带超时的阻塞取数，不用在队列空的时候疯狂轮询。

修正后的可运行参考代码：

import queue
import random
from multiprocessing import Manager, Pool


def consumer(q):
    while True:
        try:
            # 带1秒超时的阻塞取数，避免无限空转占满CPU
            res = q.get(timeout=1)
            if res is None:
                break
            print(f'Consume {res}')
        except queue.Empty:
            # 超时未取到新任务就继续循环，不退出
            continue


def producer(q, food):
    for i in range(2):
        res = f'{food} {i}'
        print(f'Produce {res}')
        q.put(res)


if __name__ == "__main__":
    CONSUMER_TOTAL = 3
    PRODUCER_TOTAL = 2
    food_catalog = ['apple', 'banana', 'melon', 'salad']
    
    with Pool() as pool:
        task_queue = Manager().Queue()
        # 提前提交所有消费者异步任务
        consumer_handles = [pool.apply_async(consumer, args=(task_queue,)) for _ in range(CONSUMER_TOTAL)]
        # 提交所有生产者异步任务
        producer_handles = [
            pool.apply_async(producer, args=(task_queue, random.choice(food_catalog)))
            for _ in range(PRODUCER_TOTAL)
        ]
        # 等待所有生产者完成全部生产动作
        for handle in producer_handles:
            handle.wait()
        # 按消费者总数发对应数量的退出哨兵
        for _ in range(CONSUMER_TOTAL):
            task_queue.put(None)
        # 等待所有消费者处理完剩余任务并退出
        for handle in consumer_handles:
            handle.wait()
        pool.close()
        pool.join()

• 如果你需要的是长期常驻的消费者进程（比如持续从队列拉取任务执行，任务结束后消费者不退出，等待下一波任务），更推荐直接用multiprocessing.Process搭配JoinableQueue实现：可以完全掌控队列的任务计数、完成信号、消费者生命周期，没有Manager代理的额外性能损耗，也不需要适配Pool的任务调度逻辑。
• 如果你的生产、消费都是短周期的一次性任务，用Pool提交异步任务、主进程统一管理哨兵的方案完全可以正常运行，不需要硬切到原生Process的实现方式。

内容的提问来源于stack exchange，提问作者funkid