如何使用Python subprocess.Popen捕获向stdin写入的子进程全部输出

报错原因 #

你遇到的io.UnsupportedOperation: read错误是subprocess的默认设计导致的：

• 当你给stdin参数传subprocess.PIPE时，父进程侧拿到的proc.stdin是只写流，设计用途是父进程向子进程的标准输入发送数据，本身就没有读权限，所以调用read()直接抛出异常。
• 默认情况下subprocess给子进程创建的stdin文件描述符是只读模式，普通子进程本身也无法直接往自己的stdin写入内容，你遇到的特殊程序是因为它主动重新打开了stdin的可写权限，或者直接操作了对应的文件描述符。

解决方案 #

要捕获子进程写入自身stdin的内容，我们需要手动创建全双工的通信通道替换默认的半双工管道，这里用Unix域套接字对（socket.socketpair）实现，它支持双向读写，完全适配这种场景。

基础实现代码 #

import subprocess
import socket
import os
import threading

# 创建全双工socket对，父进程用parent_sock，子进程用child_sock
parent_sock, child_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)

proc = subprocess.Popen(
    ('/usr/bin/modulecmd', 'python', 'help', 'null'),
    shell=False,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    stdin=child_sock.fileno(), # 把子进程端的socket作为子进程的stdin
    close_fds=True
)
# 子进程启动后关闭父进程侧持有的child_sock
child_sock.close()

# 用线程分别读取三个流，避免缓冲区满导致的死锁
stdout_buf, stderr_buf, stdin_write_buf = [], [], []

def read_stream(stream, buf):
    buf.append(stream.read())

def read_sock(sock, buf):
    while True:
        data = sock.recv(4096)
        if not data:
            break
        buf.append(data)

t1 = threading.Thread(target=read_stream, args=(proc.stdout, stdout_buf))
t2 = threading.Thread(target=read_stream, args=(proc.stderr, stderr_buf))
t3 = threading.Thread(target=read_sock, args=(parent_sock, stdin_write_buf))

t1.start()
t2.start()
t3.start()

t1.join()
t2.join()
t3.join()

proc.wait()

# 拿到三个流的字节内容
stdout = stdout_buf[0]
stderr = stderr_buf[0]
child_written_to_stdin = b''.join(stdin_write_buf)

# 用完关闭资源
parent_sock.close()

继承subprocess.Popen的封装实现 #

如果要封装成可复用的自定义Popen子类，可以参考如下实现：

import subprocess
import socket
import threading

class BiStdinPopen(subprocess.Popen):
    def __init__(self, args, **kwargs):
        # 预先创建全双工socket对替换默认stdin管道
        self.parent_stdin_sock, child_stdin_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
        kwargs['stdin'] = child_stdin_sock.fileno()
        super().__init__(args, **kwargs)
        # 关闭父进程侧持有子进程端socket句柄
        child_stdin_sock.close()
        self._stdin_buf = []
        self._read_thread = self._start_stdin_read_thread()

    def _start_stdin_read_thread(self):
        def _read_worker():
            while True:
                try:
                    data = self.parent_stdin_sock.recv(4096)
                    if not data:
                        break
                    self._stdin_buf.append(data)
                except (ConnectionResetError, BrokenPipeError):
                    break
        t = threading.Thread(target=_read_worker, daemon=True)
        t.start()
        return t

    def read_stdin_write(self) -> bytes:
        """读取子进程写入自身stdin的所有内容"""
        self._read_thread.join()
        return b''.join(self._stdin_buf)

    def __del__(self):
        try:
            self.parent_stdin_sock.close()
        except Exception:
            pass

使用方式：

proc = BiStdinPopen(
    ('/usr/bin/modulecmd', 'python', 'help', 'null'),
    shell=False,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)
stdout = proc.stdout.read()
stderr = proc.stderr.read()
child_stdin_content = proc.read_stdin_write()
proc.wait()

注意事项 #

• 全双工socket对同时支持父进程向子进程发输入、读取子进程写入stdin的内容，如果你需要同时做这两个操作，建议加锁避免读写冲突。
• 用独立线程读取不同流是必须的，否则如果子进程同时往多个流写大量内容，会因为操作系统管道缓冲区满导致进程死锁。

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