多进程（>2）场景下如何正确使用Pipe？以单生产者多消费者为例

嘿，这个问题其实戳中了Windows和Linux多进程模型的核心差异，以及Pipe在多消费者场景下的使用误区。让我先帮你分析问题根源，再给出靠谱的修正方案。

问题根源：为什么移除sleep后Linux会出错？ #

首先得明确两个系统的多进程创建方式差异：

• Windows：默认用spawn模式创建进程，子进程会重新启动Python解释器，复制父进程的资源，每个子进程的Pipe句柄是完全独立的。所以即使多个消费者共享同一个Pipe对象，底层的句柄各自独立，不会出现引用计数或竞争问题。
• Linux：默认用fork模式，子进程会直接继承父进程的文件描述符。这意味着两个消费者共享的是同一个Pipe读端的文件描述符——当生产者快速发送完数据并关闭写端后，数据会被其中一个消费者抢先读完，另一个消费者在recv时，虽然写端已经关闭，但因为读端的引用计数还没归零（另一个消费者还持有它），可能会出现无意义的阻塞，或者触发异常（具体取决于系统调度）。而sleep语句的存在，相当于给了消费者足够的时间交替读取数据，掩盖了这个问题。

另外，原代码还有一个隐藏问题：Pipe是半双工的，多个进程共享同一个读端时，数据会被随机分配给任意一个消费者，而且没有同步机制，很容易出现数据竞争或读取异常。

修正方案：两种靠谱的实现方式 #

方案一：使用multiprocessing.Queue（推荐） #

Queue是Python多进程库专门为多生产者多消费者场景设计的，底层基于Pipe和锁实现，自带进程安全的同步机制，完全不用自己处理Pipe的句柄问题。修改后的代码如下：

import multiprocessing, time

def consumer(queue, id):
    while True:
        try:
            # 使用get()方法，设置timeout避免死锁
            item = queue.get(block=True, timeout=2)
        except multiprocessing.queues.Empty:
            break
        print("%s consume：%s" % (id, item))
        # 移除sleep也不会有问题
        # time.sleep(3)
    print('Consumer done')

def producer(sequence, queue):
    for item in sequence:
        print('produce：', item)
        queue.put(item)
        time.sleep(1)

if __name__ == '__main__':
    # 创建Queue，默认大小无限制
    queue = multiprocessing.Queue()
    
    # 创建两个消费者进程
    cons_p1 = multiprocessing.Process(target=consumer, args=(queue, 1))
    cons_p1.start()
    cons_p2 = multiprocessing.Process(target=consumer, args=(queue, 2))
    cons_p2.start()
    
    sequence = [i for i in range(10)]
    producer(sequence, queue)
    
    # 等待生产者发送完所有数据，再关闭Queue
    time.sleep(2)
    queue.close()
    queue.join_thread()
    
    cons_p1.join()
    cons_p2.join()

方案二：为每个消费者创建独立的Pipe（适合广播数据场景） #

如果你确实需要用Pipe，而且希望每个消费者都能收到所有生产者的数据（广播模式），可以为每个消费者单独创建一个Pipe，生产者向每个Pipe的写端发送数据：

import multiprocessing, time

def consumer(pipe, id):
    output_p, input_p = pipe
    input_p.close()
    while True:
        try:
            item = output_p.recv()
            if item is None:  # 用None作为结束信号
                break
            print("%s consume：%s" % (id, item))
            # time.sleep(3)
        except EOFError:
            break
    print('Consumer done')

def producer(sequence, pipes):
    for item in sequence:
        print('produce：', item)
        # 向每个Pipe的写端发送数据
        for input_p in pipes:
            input_p.send(item)
        time.sleep(1)
    # 发送结束信号
    for input_p in pipes:
        input_p.send(None)

if __name__ == '__main__':
    pipes = []
    consumers = []
    
    # 为每个消费者创建独立的Pipe
    for i in range(2):
        output_p, input_p = multiprocessing.Pipe()
        pipes.append(input_p)
        cons_p = multiprocessing.Process(target=consumer, args=((output_p, input_p), i+1))
        cons_p.start()
        consumers.append(cons_p)
    
    sequence = [i for i in range(10)]
    producer(sequence, pipes)
    
    # 关闭所有写端
    for input_p in pipes:
        input_p.close()
    
    for cons_p in consumers:
        cons_p.join()

总结 #

• 如果是负载均衡式的单生产者多消费者（每个数据被一个消费者处理），优先用multiprocessing.Queue，它是最安全且易用的选择。
• 如果需要广播数据（每个消费者都收到所有数据），可以为每个消费者创建独立的Pipe。
• 避免让多个进程共享同一个Pipe的读端或写端，尤其是在Linux的fork模式下，很容易引发句柄引用计数问题和数据竞争。

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