Python使用os.fork创建4个无限进程通过os.pipe()通信的问题求助

现有代码的核心错误 #

• 对管道机制理解存在偏差：原生os.pipe()创建的是单向字节流，数据被读走就会从管道中消失，单根管道无法同时给两个读者发送同一份数据，想让B、C都读取A写入的同一条管道，必然会出现两个进程竞争抢数据、丢数的问题。
• 读写API使用错误：无参数的read()、不带长度参数的os.read()会一直阻塞到关联管道的所有写端全部关闭才返回，常驻无限进程的写端永远不会主动关闭，这行代码会永久卡住，后续逻辑根本不会执行。
• 管道描述符泄漏：fork之后每个进程都会持有所有管道的读/写端，没有及时关闭当前进程不用的端，会导致读操作永远等不到EOF，出现无意义阻塞。
• 未处理消息边界：管道是无边界字节流，直接把数字转字符串写入，读端无法判断一条消息的长度，会出现粘包、读取半条数据的问题。

可运行实现方案 #

拓扑说明 #

因为单管道不支持一对多广播，要实现A生成的数据同时同步给B、C，最简单的方案是A生成数据后同时写两根管道分别对接B、C；如果必须严格满足“A只写入一根指定管道”的要求，只需要额外加一个10行代码的轻量分发进程，读A的单管道数据后复制两份转发给B、C即可，其余业务逻辑完全一致。
所有管道在fork前统一创建，每个子进程启动后第一时间关闭自己用不到的管道端，避免描述符泄漏；主进程注册SIGINT信号处理，收到Ctrl+C时统一终止所有子进程、回收资源，避免僵尸进程。每个业务进程跑无限循环，用换行符做消息分隔符，配合行缓冲和readline()做实时读写，每次读一行刚好是一个完整数字/总和，不会出现全量阻塞。

import os
import sys
import random
import signal
import time

# 全局存储子进程pid，主进程收到SIGINT时统一回收
child_pids = []

def sigint_handler(signum, frame):
    for pid in child_pids:
        try:
            os.kill(pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
    # 等待所有子进程退出，避免僵尸进程
    for _ in range(len(child_pids)):
        os.wait()
    sys.exit(0)

if __name__ == "__main__":
    # fork前创建所有需要的管道
    # 管道ab: A写，B读
    ab_r, ab_w = os.pipe()
    # 管道ac: A写，C读
    ac_r, ac_w = os.pipe()
    # 管道cd: C写，D读
    cd_r, cd_w = os.pipe()

    # 注册SIGINT信号处理
    signal.signal(signal.SIGINT, sigint_handler)

    # 创建进程A：持续生成随机数
    pid_a = os.fork()
    if pid_a == 0:
        # 关闭当前进程不用的所有管道端
        os.close(ab_r)
        os.close(ac_r)
        os.close(cd_r)
        os.close(cd_w)
        # 行缓冲模式打开写端，写一行立刻刷新到管道
        w_ab = os.fdopen(ab_w, 'w', buffering=1)
        w_ac = os.fdopen(ac_w, 'w', buffering=1)
        while True:
            num = random.randint(1, 1000)
            # 加换行作为消息边界，同时写给B、C
            w_ab.write(f"{num}\n")
            w_ac.write(f"{num}\n")
            time.sleep(0.5) # 控制生成速度，避免占满CPU
    child_pids.append(pid_a)

    # 创建进程B：读取数字写入文件
    pid_b = os.fork()
    if pid_b == 0:
        os.close(ab_w)
        os.close(ac_r)
        os.close(ac_w)
        os.close(cd_r)
        os.close(cd_w)
        r_ab = os.fdopen(ab_r, 'r')
        # 追加模式打开存储文件，行缓冲保证实时落盘
        with open("num_storage.txt", "a", buffering=1) as f:
            while True:
                line = r_ab.readline()
                if not line:
                    break
                num = int(line.strip())
                f.write(f"{num}\n")
    child_pids.append(pid_b)

    # 创建进程C：读取数字累加，将总和写入管道给D
    pid_c = os.fork()
    if pid_c == 0:
        os.close(ab_r)
        os.close(ab_w)
        os.close(ac_w)
        os.close(cd_r)
        r_ac = os.fdopen(ac_r, 'r')
        w_cd = os.fdopen(cd_w, 'w', buffering=1)
        num_arr = []
        total = 0
        while True:
            line = r_ac.readline()
            if not line:
                break
            num = int(line.strip())
            num_arr.append(num)
            total += num
            # 每次更新总和后立刻写给D
            w_cd.write(f"{total}\n")
    child_pids.append(pid_c)

    # 创建进程D：读取总和打印输出
    pid_d = os.fork()
    if pid_d == 0:
        os.close(ab_r)
        os.close(ab_w)
        os.close(ac_r)
        os.close(ac_w)
        os.close(cd_w)
        r_cd = os.fdopen(cd_r, 'r')
        while True:
            line = r_cd.readline()
            if not line:
                break
            total = int(line.strip())
            print(f"当前所有数字累加总和: {total}", flush=True)
    child_pids.append(pid_d)

    # 主进程关闭所有自己不用的管道端，常驻等待信号
    os.close(ab_r)
    os.close(ab_w)
    os.close(ac_r)
    os.close(ac_w)
    os.close(cd_r)
    os.close(cd_w)
    while True:
        time.sleep(1)

关键注意点 #

• 不要用底层os.write直接写字符串，该方法要求传入bytes类型，用os.fdopen包装成文件对象后直接调用write()写字符串即可，配合行缓冲可以自动处理刷新逻辑，避免数据卡在缓冲区不发送。
• 所有管道端必须按需关闭：只要还有一个进程持有某根管道的写端，该管道的读端read类操作就永远不会返回EOF，会出现意外阻塞。
• 常驻进程的管道读写不要用全量read()方法，该方法仅适合一次性传输全量数据的场景，实时循环读写要配合明确的消息分隔符，用readline()或者指定固定读取长度的os.read(fd, size)实现。

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