继承subprocess.Popen类实时读写STDOUT/STDERR阻塞问题求助

你的代码核心问题在于**readline()是阻塞式调用**：当stderr没有新数据输出时，self.stderr.readline()会一直挂起等待，不会返回空字符串，直到stderr流被子进程关闭（即子进程结束）。这直接导致循环卡在stderr读取步骤，再也无法回到stdout的读取逻辑，所以后续所有STDOUT数据完全丢失。

哪怕将stderr处理放到单独线程，若仍使用阻塞的readline()，线程会因等待stderr数据而挂起，但正常情况下不会导致整个进程停滞——大概率是你主线程的等待逻辑存在问题，但本质还是阻塞读取的锅。

方案1：独立线程处理每个流（推荐） #

给stdout和stderr各自分配独立线程，每个线程只负责读取自己的流，互不干扰。利用iter(readline, b'')的特性：当流关闭时，readline()会返回空字节串，循环自动终止。

import threading
import subprocess

class MyPopen(subprocess.Popen):
    def __init__(self, *args, errHandling=subprocess.PIPE, **kwargs):
        # 强制stdout为PIPE，stderr根据配置设置
        kwargs['stdout'] = subprocess.PIPE
        kwargs['stderr'] = subprocess.PIPE if errHandling == subprocess.PIPE else kwargs.get('stderr')
        self.errHandling = errHandling
        super().__init__(*args, **kwargs)
        
        # 启动stdout处理线程
        self.stdout_thread = threading.Thread(target=self._handle_stdout, daemon=True)
        self.stdout_thread.start()
        
        # 启动stderr处理线程（若需要）
        if self.errHandling == subprocess.PIPE:
            self.stderr_thread = threading.Thread(target=self._handle_stderr, daemon=True)
            self.stderr_thread.start()
    
    def _handle_stdout(self):
        # iter(读取函数, 终止值)：当readline返回b''时，流已关闭，循环结束
        for line in iter(self.stdout.readline, b''):
            if line:
                # 解码为字符串，编码根据子进程实际输出调整
                self.processOutput(line.rstrip().decode('utf-8', errors='ignore'))
        self.stdout.close()
    
    def _handle_stderr(self):
        for line in iter(self.stderr.readline, b''):
            if line:
                self.processError(line.rstrip().decode('utf-8', errors='ignore'))
        self.stderr.close()
    
    def processOutput(self, output):
        # 自定义处理：打印+写入日志
        print(f"[STDOUT] {output}")
        with open('stdout.log', 'a', encoding='utf-8') as f:
            f.write(f"{output}\n")
    
    def processError(self, error):
        # 自定义处理：打印+写入日志
        print(f"[STDERR] {error}")
        with open('stderr.log', 'a', encoding='utf-8') as f:
            f.write(f"{error}\n")
    
    def isRunning(self):
        # 检查子进程是否存活
        return self.poll() is None

方案2：非阻塞IO读取（单线程） #

通过系统API将文件描述符设置为非阻塞模式，读取时捕获BlockingIOError（无数据时抛出），避免循环阻塞。

import os
import subprocess

class MyPopen(subprocess.Popen):
    def __init__(self, *args, errHandling=subprocess.PIPE, **kwargs):
        kwargs['stdout'] = subprocess.PIPE
        kwargs['stderr'] = subprocess.PIPE if errHandling == subprocess.PIPE else kwargs.get('stderr')
        self.errHandling = errHandling
        super().__init__(*args, **kwargs)
        
        # 给stdout和stderr设置非阻塞模式
        self._set_non_blocking(self.stdout.fileno())
        if self.errHandling == subprocess.PIPE:
            self._set_non_blocking(self.stderr.fileno())
    
    def _set_non_blocking(self, fd):
        if os.name == 'nt':
            # Windows平台使用msvcrt设置非阻塞
            import msvcrt
            msvcrt.setmode(fd, os.O_BINARY | os.O_NONBLOCK)
        else:
            # Linux/macOS使用fcntl设置非阻塞
            import fcntl
            flags = fcntl.fcntl(fd, fcntl.F_GETFL)
            fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
    
    def read_outputs(self):
        while self.isRunning():
            # 读取stdout
            try:
                line = self.stdout.readline()
                if line:
                    self.processOutput(line.rstrip().decode('utf-8', errors='ignore'))
            except BlockingIOError:
                # 无数据时跳过
                pass
            
            # 读取stderr
            if self.errHandling == subprocess.PIPE:
                try:
                    line = self.stderr.readline()
                    if line:
                        self.processError(line.rstrip().decode('utf-8', errors='ignore'))
                except BlockingIOError:
                    pass
    
    # processOutput、processError、isRunning同方案1

• 编码匹配：子进程输出的编码可能和系统有关（Windows常为GBK，类Unix为UTF-8），需根据实际情况调整decode()的编码参数，或让子进程强制输出UTF-8。
• 线程守护：设置daemon=True可让线程随主进程退出自动终止，避免残留资源。
• 流关闭：处理完流后主动关闭，防止资源泄漏。

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