如何实现每个值可被访问两次的multiprocessing Queue？

原生multiprocessing.Queue是独占式消费，每个元素只能被一个进程取走，所以你当前的代码里两个进程会瓜分数据。要实现每个数据被两个进程都获取，你可以用转发进程+独立子队列的方案，完美适配你提到的两个复杂场景（数据节奏不定、生成速度快于处理速度）。

具体实现思路 #

1. 生成进程只往一个主队列推送数据
2. 新增一个转发进程，从主队列取出数据后，复制一份分别推送给两个子队列（对应Process1和Process2）
3. Process1和Process2各自监听自己的子队列，独立处理全部数据，子队列会自动缓存未处理的数据，应对生成速度过快的情况

修改后的完整代码 #

import multiprocessing as mp
import numpy as np
import time
import random

class ForwardProcess(mp.Process):
    def __init__(self, main_queue, queue1, queue2):
        super().__init__()
        self.main_queue = main_queue
        self.queue1 = queue1
        self.queue2 = queue2

    def run(self):
        while True:
            try:
                data = self.main_queue.get()
                # 收到终止信号则转发后退出
                if data is None:
                    self.queue1.put(None)
                    self.queue2.put(None)
                    self.main_queue.task_done()
                    break
                # numpy数组需复制确保进程间数据独立
                self.queue1.put(data.copy() if isinstance(data, np.ndarray) else data)
                self.queue2.put(data.copy() if isinstance(data, np.ndarray) else data)
                self.main_queue.task_done()
            except Exception as e:
                print(f"转发进程出错: {e}")
                break

class Process1(mp.Process):
    def __init__(self, data_queue):
        super().__init__()
        self.data_queue = data_queue

    def run(self):
        while True:
            data = self.data_queue.get()
            if data is None:
                self.data_queue.task_done()
                break
            # Process1的处理逻辑
            print(f"Process1处理数据，形状: {data.shape}")
            self.data_queue.task_done()

class Process2(mp.Process):
    def __init__(self, data_queue):
        super().__init__()
        self.data_queue = data_queue

    def run(self):
        while True:
            data = self.data_queue.get()
            if data is None:
                self.data_queue.task_done()
                break
            # Process2的处理逻辑
            print(f"Process2处理数据，均值: {round(np.mean(data), 4)}")
            self.data_queue.task_done()

if __name__ == "__main__":
    # 使用JoinableQueue方便等待任务处理完成
    main_queue = mp.JoinableQueue()
    queue_p1 = mp.JoinableQueue()
    queue_p2 = mp.JoinableQueue()

    # 启动进程
    forwarder = ForwardProcess(main_queue, queue_p1, queue_p2)
    p1 = Process1(queue_p1)
    p2 = Process2(queue_p2)

    forwarder.start()
    p1.start()
    p2.start()

    # 模拟外部不定节奏的数据生成
    try:
        while True:
            batch_size = random.randint(1, 10)
            for _ in range(batch_size):
                main_queue.put(np.random.rand(1000))
            wait_time = random.uniform(0.1, 5)
            time.sleep(wait_time)
    except KeyboardInterrupt:
        # 发送终止信号并等待所有任务完成
        main_queue.put(None)
        main_queue.join()
        queue_p1.join()
        queue_p2.join()
        forwarder.join()
        p1.join()
        p2.join()
        print("所有进程已终止")

关键说明 #

• numpy数组复制：因为numpy数组在进程间传递时默认共享内存，用copy()确保两个进程拿到独立副本，避免处理时互相干扰
• 终止信号：通过推送None实现所有进程优雅退出，适配实际场景的停止需求
• 缓存能力：队列会自动缓存未处理的数据，即使生成速度远超处理速度，也不会丢失数据
• 不定节奏适配：队列的异步特性天然兼容数据时快时慢的场景，无需额外逻辑

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