Python多进程共享Queue 多消费者获取相同元素问题求解

你当前的代码有两个核心认知偏差，直接导致了异常：

• multiprocessing.Queue的设计逻辑是点对点任务分发队列，遵循「取出即删除」规则：同一份元素只要被任意一个进程通过get()取走，队列中就不再留存该元素，多个消费者共享同一个队列时本质是在争抢队列内的元素，自然不可能每个消费者都拿到全量的相同数据，这就是你看到只有一个消费者打印输出的核心原因。
• 终止逻辑错误：你仅向队列中放入了1个None作为终止信号，这个信号只会被其中一个消费者抢到，另一个消费者会永远阻塞在q.get()调用上等待不存在的新数据，整个脚本自然无法正常终止。

注意：就算你往队列里放和消费者数量一致的None终止标记，也解决不了元素争抢的问题，依然无法实现「所有消费者拿到完全相同元素」的需求。

最稳定、无额外依赖的实现方式是为每个消费者分配独立的专属队列，生产者生产数据时，将同一份数据依次投递到所有消费者的队列中即可。
修正后的可运行代码如下：

from multiprocessing import Process, Queue

# 消费者总数量
CONSUMER_COUNT = 2

def producer(queue_list):
    # 生产业务数据
    for data in range(10):
        # 把同一份数据投递到每个消费者的专属队列
        for q in queue_list:
            q.put(data)
    # 给每个消费者单独发送终止标记
    for q in queue_list:
        q.put(None)

def consumer1(q):
    while True:
        data = q.get()
        if data is None:
            break
        print(f"[consumer1] 取到数据: {data}")

def consumer2(q):
    while True:
        data = q.get()
        if data is None:
            break
        print(f"[consumer2] 取到数据: {data}")

def main():
    # 为每个消费者初始化独立队列
    queues = [Queue() for _ in range(CONSUMER_COUNT)]
    prod_process = Process(target=producer, args=(queues,))
    c1_process = Process(target=consumer1, args=(queues[0],))
    c2_process = Process(target=consumer2, args=(queues[1],))

    # 启动所有进程
    prod_process.start()
    c1_process.start()
    c2_process.start()

    # 等待所有进程执行完毕
    prod_process.join()
    c1_process.join()
    c2_process.join()

if __name__ == '__main__':
    main()

方案说明 #

• 该方案下每个消费者读取自己的专属队列，不会出现元素争抢，所有消费者都能拿到完整的0-9全量数据，且每个消费者都能收到专属的终止标记，进程可以正常退出。
• 如果单条数据内存占用极大、消费者数量较多，重复投递数据会占用过多内存，可以将原始数据存入多进程共享内存，队列中仅传递数据的索引位置即可，通用场景下直接使用独立队列是实现成本最低、bug率最低的方案。
• 不要尝试用「消费者取到数据后再塞回队列」的方式在单队列上实现广播，这类写法会引发死锁、数据顺序混乱问题，生产环境绝对不要用。

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