Python2.7中多Popen管道子进程的全stderr捕获方案问询

这个问题我之前也碰到过，串联subprocess管道时想捕获每个进程的stderr确实有点坑，先给你理清为什么你之前试的逐个调用communicate()不行，再给你两个靠谱的解决方案：

首先说为什么编辑1里的方案走不通：当你把p1的stdout连到p2的stdin后，p2会持续读取p1的输出，这时候如果对p1调用communicate()，会试图关闭p1的stdin（虽然这里我们没用到），更关键的是，这会干扰p2的读取流程，导致管道异常；而且进程结束后再调用communicate()也会直接报错，所以这个思路肯定行不通。

下面给你两种可行的实现方式：

方案一：用线程实时读取每个进程的stderr #

因为subprocess的管道缓冲区如果被占满，会导致进程阻塞挂起，所以我们可以为每个进程的stderr单独开一个线程，实时读取输出并保存到缓冲区里，这样就能在Python内部捕获所有进程的错误输出，不用依赖外部文件。

示例代码：

import subprocess as sp
import shlex
import threading

def read_stderr(process, stderr_buffer):
    """读取进程的stderr并写入缓冲区"""
    while True:
        line = process.stderr.readline()
        if not line:
            break
        stderr_buffer.append(line)

# 替换成你的实际命令列表
cmds = ["echo 'test cmd1' && echo 'error1' >&2", "grep 'test' && echo 'error2' >&2", "wc -l && echo 'error3' >&2"]

processes = []
stderr_buffers = []
threads = []

# 启动第一个进程
p1 = sp.Popen(shlex.split(cmds[0]), stdout=sp.PIPE, stderr=sp.PIPE)
processes.append(p1)
buf1 = []
stderr_buffers.append(buf1)
t1 = threading.Thread(target=read_stderr, args=(p1, buf1))
t1.daemon = True
t1.start()
threads.append(t1)

# 启动后续串联的进程
for i in range(1, len(cmds)):
    prev_process = processes[i-1]
    current_process = sp.Popen(
        shlex.split(cmds[i]),
        stdin=prev_process.stdout,
        stdout=sp.PIPE,
        stderr=sp.PIPE
    )
    processes.append(current_process)
    current_buf = []
    stderr_buffers.append(current_buf)
    current_thread = threading.Thread(target=read_stderr, args=(current_process, current_buf))
    current_thread.daemon = True
    current_thread.start()
    threads.append(current_thread)
    # 关闭前一个进程的stdout，让它在输出结束后能正常退出
    prev_process.stdout.close()

# 等待最后一个进程完成，获取最终的标准输出
final_stdout, _ = processes[-1].communicate()

# 等待所有线程结束，确保所有stderr都被读取完毕
for thread in threads:
    thread.join()

# 输出每个进程的stderr内容
for idx, stderr_lines in enumerate(stderr_buffers):
    print(f"=== 进程{idx+1}的stderr ===")
    print(''.join(stderr_lines))
print(f"=== 最终进程的stdout ===")
print(final_stdout)

这个方案的优势是可以实时处理每个进程的错误输出，比如输出到日志或者实时展示，而且不需要依赖外部文件；需要注意的是必须启动下一个进程后关闭前一个进程的stdout，否则前一个进程可能会因为管道未关闭而无法正常退出。

方案二：重定向stderr到临时文件（你提到的可靠方案） #

如果不需要实时处理stderr，只是想在所有进程结束后一次性收集，那用临时文件的方式更简单，也不容易出问题，适合场景比较简单的情况：

import subprocess as sp
import shlex
import tempfile
import os

# 替换成你的实际命令列表
cmds = ["echo 'test cmd1' && echo 'error1' >&2", "grep 'test' && echo 'error2' >&2", "wc -l && echo 'error3' >&2"]

processes = []
stderr_temp_files = []

# 启动第一个进程，stderr重定向到临时文件
with tempfile.NamedTemporaryFile(delete=False) as tf:
    stderr_temp_files.append(tf.name)
p1 = sp.Popen(shlex.split(cmds[0]), stdout=sp.PIPE, stderr=open(stderr_temp_files[0], 'w'))
processes.append(p1)

# 启动后续串联的进程
for i in range(1, len(cmds)):
    with tempfile.NamedTemporaryFile(delete=False) as tf:
        stderr_temp_files.append(tf.name)
    prev_process = processes[i-1]
    current_process = sp.Popen(
        shlex.split(cmds[i]),
        stdin=prev_process.stdout,
        stdout=sp.PIPE,
        stderr=open(stderr_temp_files[i], 'w')
    )
    processes.append(current_process)
    prev_process.stdout.close()

# 等待所有进程执行完成
for p in processes:
    p.wait()

# 读取并输出每个进程的stderr内容
for idx, file_path in enumerate(stderr_temp_files):
    print(f"=== 进程{idx+1}的stderr ===")
    with open(file_path, 'r') as f:
        print(f.read())
    # 清理临时文件
    os.unlink(file_path)

# 获取最终进程的stdout（如果需要）
final_stdout = processes[-1].stdout.read()
print(f"=== 最终进程的stdout ===")
print(final_stdout)

这个方案的好处是代码简洁，不需要处理线程逻辑，可靠性很高；缺点是无法实时获取stderr，必须等所有进程都结束后才能读取。

总结一下：如果需要实时处理每个进程的错误输出，优先选线程方案；如果只需要事后收集，临时文件方案更省心。

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