如何正确中断阻塞读取子进程输出的Python线程？

你的问题核心在于：直接关闭子进程的stdout管道会触发底层内存错误，而且close()操作会因为管道的阻塞状态挂起，直到read()抛出异常。这不是处理这类场景的正确方式，我们需要让子进程和读取线程都能优雅退出，而非靠强制关闭管道引发异常。

问题根源分析 #

当线程1执行p.stdout.read()时，它处于阻塞状态，等待子进程输出数据。此时线程2调用self.closer.close()，直接关闭管道一端，但子进程可能仍在运行，这会导致管道内部状态异常，进而触发ValueError。同时，close()操作本身可能因管道缓冲未处理完成而阻塞，直到read()抛出异常才会返回。

正确解决方案：先终止子进程，再等待线程退出 #

最稳妥的方式是先终止子进程，让子进程自动关闭stdout管道，线程1的read()会因管道另一端关闭返回空字节，从而正常退出循环。同时，我们用线程安全的threading.Event替代普通的self.running变量，避免多线程下的可见性问题。

修改后的代码示例：

Thread 1 代码：

import threading
import subprocess

class Worker:
    def __init__(self):
        self.stop_event = threading.Event()
        self.process = None

    def run(self):
        print("Starting to read.")
        # 启动子进程，替换为你的实际命令
        self.process = subprocess.Popen(
            ["your_command_here"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True
        )
        # 循环读取输出，直到进程结束或收到停止信号
        for line in iter(self.process.stdout.readline, ''):
            if self.stop_event.is_set():
                break
            # 处理子进程输出
            print(f"Output: {line.strip()}")
        # 等待进程结束并清理资源
        self.process.wait()
        print("Finishing")

Thread 2 代码：

import time

def stop_worker(worker, worker_thread):
    print("About to close.")
    # 发送停止信号给线程1
    worker.stop_event.set()
    # 终止子进程（如果仍在运行）
    if worker.process and worker.process.poll() is None:
        worker.process.terminate()
        # 等待3秒，若进程未退出则强制杀死
        try:
            worker.process.wait(timeout=3)
        except subprocess.TimeoutExpired:
            worker.process.kill()
    # 等待线程1完全退出
    worker_thread.join()
    print("Finished")

方案有效性说明 #

1. 终止子进程：process.terminate()给子进程发送终止信号（如SIGTERM），子进程退出后会自动关闭stdout管道，线程1的readline()会读取到空字符串，正常退出循环。
2. 线程安全的停止信号：threading.Event是线程安全的，线程1可在每次读取后检查是否需要停止，即使子进程未退出，也能及时响应中断请求。
3. 避免直接关闭管道：无需手动关闭stdout，让子进程正常关闭管道，保证read()操作正常返回，不会触发异常。

备选方案：非阻塞读取（适合即时响应场景） #

若需在不终止子进程的情况下中断读取，可将stdout设为非阻塞模式，结合select模块检查数据可读性，同时定期检查停止信号。这种方式不会让线程长时间阻塞，但会增加少量CPU占用：

import os
import select
import threading
import subprocess

class NonBlockingWorker:
    def __init__(self):
        self.stop_event = threading.Event()
        self.process = None

    def run(self):
        print("Starting to read (non-blocking).")
        self.process = subprocess.Popen(
            ["your_command_here"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True
        )
        # 将stdout设为非阻塞模式
        fd = self.process.stdout.fileno()
        os.set_blocking(fd, False)
        
        while not self.stop_event.is_set():
            # 等待数据可读，超时1秒
            ready, _, _ = select.select([fd], [], [], 1)
            if ready:
                line = self.process.stdout.read()
                if not line:
                    break
                print(f"Output: {line.strip()}")
        
        self.process.terminate()
        self.process.wait()
        print("Finishing")

这个方案中，select.select()最多阻塞1秒，之后会检查停止信号，线程能及时响应中断请求，不会一直卡在read()上。

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