如何用Python连接进程stdin/stdout并实现Bash的FIFO通信流程？

好问题！其实你完全不需要手动创建FIFO文件——Python的subprocess模块可以直接帮你构建source -> processor -> sink的进程数据流管道，而且比Bash的FIFO方式更简洁、更易维护。

核心思路 #

Python的subprocess.Popen支持直接将一个进程的标准输出（stdout）连接到另一个进程的标准输入（stdin），形成链式管道，无需依赖外部的FIFO文件。下面是具体的实现示例：

示例代码 #

假设你的source、processor、sink是可执行命令（比如系统命令、自定义脚本），这里用常见的命令作为演示，你可以直接替换成自己的实际命令：

import subprocess

# 替换成你实际的source/processor/sink命令
source_cmd = ["echo", "hello from the source process"]  # 示例：输出文本
processor_cmd = ["tr", "a-z", "A-Z"]  # 示例：将小写转大写
sink_cmd = ["cat"]  # 示例：打印输出到终端

# 1. 启动processor进程，预留stdin和stdout用于管道连接
processor = subprocess.Popen(
    processor_cmd,
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True  # 处理文本数据；如果是二进制数据，去掉这个参数，用bytes交互
)

# 2. 启动source，将其stdout直接接入processor的stdin
source = subprocess.Popen(
    source_cmd,
    stdout=processor.stdin,
    text=True
)

# 3. 启动sink，将processor的stdout接入sink的stdin
sink = subprocess.Popen(
    sink_cmd,
    stdin=processor.stdout,
    text=True
)

# 关键：关闭不需要的管道端点，避免进程因等待EOF而挂起
source.stdout.close()       # source结束后，让processor收到EOF
processor.stdin.close()     # 断开与source的连接，防止processor等待额外输入
processor.stdout.close()    # processor结束后，让sink收到EOF

# 等待所有进程执行完成，并检查返回状态
source_exit_code = source.wait()
processor_exit_code = processor.wait()
sink_exit_code = sink.wait()

# 可选：检查进程是否正常退出
if source_exit_code != 0:
    print(f"⚠️ Source进程异常退出，返回码: {source_exit_code}")
if processor_exit_code != 0:
    print(f"⚠️ Processor进程异常退出，返回码: {processor_exit_code}")
if sink_exit_code != 0:
    print(f"⚠️ Sink进程异常退出，返回码: {sink_exit_code}")

关键细节解释 #

• 管道连接：通过stdout=processor.stdin这样的参数，直接将前一个进程的输出绑定到下一个进程的输入，Python会自动管理底层的管道通信。
• 关闭管道：必须手动关闭不再使用的管道端点，否则进程可能会一直等待输入/输出的EOF信号，导致挂起。
• 文本/二进制模式：text=True会让进程的输入输出以字符串形式处理；如果需要处理二进制数据（比如图片、音频），去掉这个参数即可，此时需要用bytes类型进行交互。

自定义脚本适配 #

如果你的source、processor、sink是Python脚本，只需要把命令换成对应的Python调用即可，比如：

source_cmd = ["python3", "./source.py"]
processor_cmd = ["python3", "./processor.py"]
sink_cmd = ["python3", "./sink.py"]

这种方式相比Bash的FIFO方案，不需要手动创建和清理FIFO文件，也避免了FIFO可能带来的权限或残留问题，代码的可读性和可维护性也更高。

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