如何通过Python subprocess模块获取子进程的当前输出行

你的核心问题是readline()只会在遇到换行符时返回数据，但input()的提示（比如请输入：）通常不带换行，导致无法实时捕获。下面是具体的解决方案：

关键调整点 #

• 放弃按行读取的方式，改用直接读取字节流，不依赖换行符触发返回
• 配置子进程为无缓冲模式，确保输出实时推送
• 用线程分别处理stdout和stderr，避免单线程阻塞

完整代码示例 #

1. 子进程启动配置 #

import subprocess
import os
import sys
import threading

# 启动子进程：设置stdin管道用于交互，无缓冲，字节模式
process = subprocess.Popen(
    ['python', '-u', 'test.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    stdin=subprocess.PIPE,
    bufsize=0,  # 禁用缓冲
    text=False  # 使用字节流，后续手动解码
)

2. 实时读取输出的线程函数 #

def read_stream(stream, stream_name):
    while True:
        try:
            # 每次读取1字节（也可以调整为更大的块如1024，平衡性能和实时性）
            data = os.read(stream.fileno(), 1)
            if not data:
                break  # 流关闭时退出循环
            # 解码为字符串，根据实际编码调整（比如gbk）
            content = data.decode('utf-8')
            print(f"{content}", end='')  # end=''避免额外换行
            sys.stdout.flush()
        except Exception:
            break

# 启动两个线程分别处理标准输出和错误输出
threading.Thread(target=read_stream, args=(process.stdout, 'stdout'), daemon=True).start()
threading.Thread(target=read_stream, args=(process.stderr, 'stderr'), daemon=True).start()

3. 处理用户输入交互 #

# 读取用户输入并发送给子进程
while True:
    try:
        user_input = input()
        # 输入exit时终止子进程
        if user_input.strip().lower() == 'exit':
            process.stdin.write((user_input + '\n').encode('utf-8'))
            process.stdin.flush()
            break
        # 发送输入给子进程，必须加换行符模拟终端输入
        process.stdin.write((user_input + '\n').encode('utf-8'))
        process.stdin.flush()
    except EOFError:
        break

# 等待子进程结束
process.wait()

测试用test.py示例 #

import time

# 无换行的输出，模拟实时进度提示
print("正在初始化...", end='', flush=True)
time.sleep(2)
print("完成！")

# 带提示的输入
user_content = input("请输入你的内容：")
print(f"你输入的内容是：{user_content}")

原理说明 #

• os.read()直接操作文件描述符，不依赖行缓冲，能捕获任何长度的输出（包括未换行的提示）
• bufsize=0和python -u确保子进程的stdout/stderr不做缓冲，实时输出数据
• 多线程处理stdout/stderr避免了单线程下读一个流阻塞另一个流的问题

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