subprocess.Popen异步流实现原理及同类BufferedReader复刻方法

关于subprocess.Popen阻塞读取的核心机制 #

你观察到的“等待新行、进程结束才退出循环”的效果，没有依赖任何subprocess模块的特殊机制，也不需要async/await支撑，完全是操作系统层面的阻塞IO自带的能力：

• subprocess.Popen创建子进程时会为标准输出创建匿名管道，p.stdout本质是绑定了这个管道读端文件描述符的BufferedReader对象
• 调用readline()时如果管道中没有可用数据，操作系统会直接将调用该方法的线程挂起，直到管道有新数据写入、或管道被关闭（子进程退出）才会唤醒线程返回结果，全程不需要Python层面做轮询、加time.sleep或者事件驱动逻辑
• 你用到的iter(p.stdout.readline, b'')是Python内置的双参数迭代器语法：会反复调用第一个可调用对象，直到返回值等于第二个哨兵值b''（管道关闭时readline会返回空字节串）就终止迭代，逻辑完全是通用语法，和subprocess无关

模拟等价的阻塞读对象非常简单 #

只要利用Python内置的同步原语就能实现完全相同的等待行为，不需要复刻BufferedReader的全部逻辑：
最常用的方案是用queue.Queue，它的get()方法天生支持阻塞等待，有新数据才返回，队列关闭时抛出异常终止迭代，效果和管道readline完全一致。

多线程转发输出的实现方案 #

管道的读操作是消费型的，数据被读取后就会从管道中移除，因此不能直接复制两个BufferedReader同时读取。正确的实现思路如下：

1. 单独启动一个专属读线程，唯一负责从原始p.stdout中逐行读取数据
2. 初始化两个线程安全的queue.Queue实例，分别对应两个消费线程
3. 读线程每拿到一行数据，就同时往两个队列中各写入一份
4. 两个消费线程分别从自己对应的队列中阻塞读取数据，Queue.get()的阻塞行为和原readline()完全等价

示例代码参考：

import subprocess
import threading
from queue import Queue

def consumer(queue, name):
    while True:
        line = queue.get()
        if line is None: # 哨兵值标记读取结束
            break
        print(f"消费者{name}拿到: {line}")

# 启动子进程
p = subprocess.Popen('some command', stdout=subprocess.PIPE)

# 初始化队列和消费者线程
q1 = Queue()
q2 = Queue()
t1 = threading.Thread(target=consumer, args=(q1, "1"))
t2 = threading.Thread(target=consumer, args=(q2, "2"))
t1.start()
t2.start()

# 读线程负责转发数据
def read_worker(p, q1, q2):
    for line in iter(p.stdout.readline, b''):
        q1.put(line)
        q2.put(line)
    # 读取结束后写入哨兵通知消费者退出
    q1.put(None)
    q2.put(None)

threading.Thread(target=read_worker, args=(p, q1, q2)).start()

# 等待所有线程结束
t1.join()
t2.join()
p.wait()

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