如何通过Python Popen读取子进程除stdout、stdin外的自定义流

subprocess默认只封装了stdin/stdout/stderr三个标准流的配置，要读取自定义编号的流，需要配合os模块的管道能力实现，核心思路是在父进程提前创建对应管道，将写端映射到子进程的3、4号文件描述符，父进程从读端读取数据即可。

完整实现代码 #

import os
import subprocess
import threading

# 1. 创建两对管道，分别对应视频流、音频流，每对管道含[读端fd, 写端fd]
r_video, w_video = os.pipe()
r_audio, w_audio = os.pipe()

def redirect_child_fds():
    # 子进程启动前执行：把管道写端映射到3、4号文件描述符
    os.dup2(w_video, 3)
    os.dup2(w_audio, 4)

# 2. 启动子进程
popen = subprocess.Popen(
    ["./test1"],
    stdout=subprocess.PIPE,
    pass_fds=(w_video, w_audio), # 把两个写端fd传递给子进程，不被默认关闭
    preexec_fn=redirect_child_fds
)

# 3. 父进程关闭不需要的管道写端，否则读端会一直等待EOF阻塞
os.close(w_video)
os.close(w_audio)

# 4. 定义各流的读取逻辑
def read_stdout():
    for line in iter(popen.stdout.readline, b""):
        print(line.decode("utf-8"), end="")

def read_video_stream():
    while True:
        # 按自己需要的缓冲区大小读取，这里示例每次读1MB
        data = os.read(r_video, 1024 * 1024)
        if not data:
            break
        # 处理视频数据的逻辑写在这里
        print(f"读取到视频数据，长度：{len(data)}")

def read_audio_stream():
    while True:
        data = os.read(r_audio, 1024 * 1024)
        if not data:
            break
        # 处理音频数据的逻辑写在这里
        print(f"读取到音频数据，长度：{len(data)}")

# 5. 开三个线程异步读取三个流，避免单线程读取时某个流缓冲区满导致子进程阻塞
threading.Thread(target=read_stdout, daemon=True).start()
threading.Thread(target=read_video_stream, daemon=True).start()
threading.Thread(target=read_audio_stream, daemon=True).start()

# 等待子进程退出
popen.wait()
# 关闭剩余的读端
os.close(r_video)
os.close(r_audio)

注意事项 #

• 该方案仅适用于Linux/Unix/macOS环境，Windows系统的文件描述符传递逻辑不同需要单独适配
• 不要给Popen设置text=True或universal_newlines=True，否则二进制的音视频数据会被强制转码导致损坏
• 必须异步读取所有流，若只读取其中一个流，其他流的缓冲区满后会导致子进程挂起无法继续运行

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