如何安全阻塞multiprocessing.Queue直至消息被另一进程接收？

问题根源 #

你的MessagePasser类存在两个核心问题：

• 单队列无方向区分：共享队列没有区分消息的发送方向，导致发送进程可能取到自己发送的消息（比如Alice发送整数后，自己的receive_message抢先取走了这个整数，后续调用sum()就会触发AttributeError）。
• 同步机制不可靠：单个事件被两个进程共享，容易出现信号混乱；同时用while self.queue.empty(): pass轮询队列状态是非原子操作，存在竞态条件，可能导致挂起或错误。

修复方案 #

我们可以给双向通信分别创建独立的通道（每个方向一个队列+一个事件），确保消息的收发严格对应，同时用队列原生的阻塞get()替代轮询，保证原子性。修改后的代码如下：

import multiprocessing
import numpy as np

class MessagePasser:
    def __init__(self):
        # 分别对应单方向的队列与同步事件
        self.queue = multiprocessing.Queue(maxsize=1)
        self.event = multiprocessing.Event()
    
    def send(self, message: np.ndarray|int) -> None:
        self.queue.put(message)
        # 阻塞直到对方接收完成并发出信号
        self.event.wait()
        self.event.clear()

    def recv(self) -> np.ndarray|int:
        # 用队列原生阻塞get替代轮询，保证原子性
        message = self.queue.get()
        # 通知发送方已完成接收
        self.event.set()
        return message

def alices_job(send_to_bob: MessagePasser, recv_from_bob: MessagePasser):
    for i in range(1000):
        array = np.random.rand(10)
        integer = 1
        send_to_bob.send(array)
        send_to_bob.send(integer)
        modified_array = recv_from_bob.recv()
        modified_array.sum()

def bobs_job(send_to_alice: MessagePasser, recv_from_alice: MessagePasser):
    for i in range(1000):
        array = recv_from_alice.recv().astype(float)
        integer = recv_from_alice.recv()
        array *= int(integer)
        send_to_alice.send(array)

if __name__ == "__main__":
    # 创建两个独立通道，分别处理Alice→Bob和Bob→Alice的消息
    alice_to_bob = MessagePasser()
    bob_to_alice = MessagePasser()
    
    alices_process = multiprocessing.Process(
        target=alices_job, 
        args=(alice_to_bob, bob_to_alice)
    )
    bobs_process = multiprocessing.Process(
        target=bobs_job, 
        args=(bob_to_alice, alice_to_bob)
    )
    
    alices_process.start()
    bobs_process.start()
    alices_process.join()
    bobs_process.join()

关键改进点 #

• 分离双向通道：用两个独立的MessagePasser实例分别处理不同方向的消息，彻底避免消息被发送方误接收。
• 原子性阻塞操作：使用队列原生的get()方法（默认阻塞）替代轮询empty()，消除竞态条件导致的异常。
• 独立同步事件：每个通道的同步事件独立，确保发送方只会等待对应接收方的确认信号，不会出现信号混乱。

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