使用Python subprocess.Popen读取stdout时陷入死锁的解决方法咨询

你的代码出现死锁，核心原因有两个：一是readline()会一直阻塞，直到有新输出或进程结束，但你的目标进程在5秒后会等待stdin输入，不会主动终止；二是你没有处理stderr管道，若进程往stderr输出内容，管道缓冲区满后会导致进程挂起，进一步加剧死锁。

下面给出三种实用解决方案，按需选择：

方案1：用select实现非阻塞实时读取 #

适合需要实时监控stdout/stderr，同时处理进程交互的场景：

import subprocess
import select

process = subprocess.Popen(
    ['python3', 'test.py'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

lines = []
while True:
    # 监听stdout和stderr，设置1秒超时避免无限阻塞
    ready, _, _ = select.select([process.stdout, process.stderr], [], [], 1)
    
    for pipe in ready:
        line = pipe.readline()
        if line:
            if pipe is process.stdout:
                lines.append(line)
                print(f"输出：{line.strip()}")
            else:
                print(f"错误：{line.strip()}")
    
    # 检查进程是否已终止
    if process.poll() is not None:
        # 读取剩余输出
        remaining_stdout = process.stdout.read()
        if remaining_stdout:
            lines.extend(remaining_stdout.splitlines(keepends=True))
        remaining_stderr = process.stderr.read()
        if remaining_stderr:
            print(f"剩余错误输出：{remaining_stderr.strip()}")
        break

# 若需要回应进程的继续询问，取消下面注释
# process.stdin.write("y\n")
# process.stdin.flush()

方案2：用communicate()安全交互并获取全部输出 #

如果你只需要给进程发送一次输入（比如回答是否继续），然后获取所有输出，communicate()是最稳妥的选择——它会自动处理stdout和stderr的读取，从根源避免死锁：

import subprocess

process = subprocess.Popen(
    ['python3', 'test.py'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 发送输入（示例为回答"y"继续），设置超时避免无限等待
stdout, stderr = process.communicate(input="y\n", timeout=10)

lines = stdout.splitlines(keepends=True)
print("所有输出：")
for line in lines:
    print(line.strip())

if stderr:
    print("\n错误输出：")
    print(stderr.strip())

方案3：用线程分离输出读取 #

适合长期运行的进程，需要在主线程处理交互，同时实时处理输出的场景：

import subprocess
import threading
import time

def read_output(pipe, callback):
    # 迭代读取直到管道关闭
    for line in iter(pipe.readline, ''):
        callback(line)
    pipe.close()

process = subprocess.Popen(
    ['python3', 'test.py'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

lines = []
# 处理stdout的回调函数
def handle_stdout(line):
    lines.append(line)
    print(f"输出：{line.strip()}")

# 处理stderr的回调函数
def handle_stderr(line):
    print(f"错误：{line.strip()}")

# 启动线程分别读取stdout和stderr
stdout_thread = threading.Thread(target=read_output, args=(process.stdout, handle_stdout))
stderr_thread = threading.Thread(target=read_output, args=(process.stderr, handle_stderr))
stdout_thread.start()
stderr_thread.start()

# 等待进程弹出询问，发送输入
time.sleep(5)
process.stdin.write("y\n")
process.stdin.flush()

# 等待线程和进程结束
stdout_thread.join()
stderr_thread.join()
process.wait()

print("\n最终收集的输出：")
for line in lines:
    print(line.strip())

方案选型建议 #

• 实时监控+交互：选select或线程方案
• 单次交互+获取全输出：选communicate()

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