Python subprocess父子进程JSON交互时如何非阻塞实时读stdout

主进程需向启动的子进程发送JSON结构化数据，子进程接收数据完成处理逻辑，同时将百分比格式的处理进度返回主进程，用于主进程更新用户界面的进度条组件。

当前主进程仅能在子进程完全执行结束后接收到子进程的全部输出，推测是stdout的read()调用阻塞导致，需求为子进程每向stdout写入一行内容，主进程即可立即读取处理该响应，无需等待子进程退出。

• 操作系统：Windows
• Python版本：3.10
• 集成开发环境：PyCharm

parent.py #

from json import dumps
import subprocess
from time import sleep

lines_to_exchange = ["this is line one", "this is line two", "this is line three", "this is line four", "this is line five"]
command = ["python", "-u", "./child.py"]
print("start process")


sub = subprocess.Popen(command, text=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

sub.stdin.write(dumps(lines_to_exchange))
sub.stdin.close()


while True:
    sleep(0.1)
    stdout = sub.stdout.read()
    print(stdout)

    if sub.poll() is not None:
        print("process completed")
        break

child.py #

from time import sleep
from json import loads
import sys
lines = loads(input())

for line in lines:
    sleep(1)
    print(line)
    sys.stdout.flush()

核心问题出在主进程的读取逻辑：无参数调用sub.stdout.read()时，该方法会持续读取stdout管道内的所有内容，直到管道被关闭（即子进程退出、stdout句柄释放）才会返回，因此哪怕子进程已经主动flush缓冲区、内容已经到达主进程侧的管道缓存，read()也不会提前返回，必须等子进程执行完毕才能拿到全部输出。

1. 将主进程的全量读取逻辑改为逐行读取，使用readline()方法替代无参数的read()。readline()只要在管道中读取到换行符，就会立即返回当前行内容，不会阻塞等待后续输出。
2. 子进程侧可以直接在print()方法中传入flush=True参数，省略手动调用sys.stdout.flush()的步骤，代码更简洁。
3. 保留启动子进程时的-u参数，该参数用于禁用Python解释器的stdout块缓冲，避免Windows环境下输出滞留缓冲区不及时发送的问题。

修复后主进程代码 #

from json import dumps
import subprocess

lines_to_exchange = ["this is line one", "this is line two", "this is line three", "this is line four", "this is line five"]
command = ["python", "-u", "./child.py"]
print("start process")


sub = subprocess.Popen(command, text=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

sub.stdin.write(dumps(lines_to_exchange))
sub.stdin.close()


while True:
    stdout_line = sub.stdout.readline()
    if stdout_line:
        print(stdout_line.strip())
    # 子进程退出且管道内无剩余内容时结束循环
    if sub.poll() is not None and not stdout_line:
        print("process completed")
        break

修复后子进程代码 #

from time import sleep
from json import loads
lines = loads(input())
total_task = len(lines)

for idx, line in enumerate(lines, 1):
    sleep(1)
    # 输出处理内容
    print(line, flush=True)
    # 如需返回进度直接输出百分比即可，主进程可逐行解析更新进度条
    # print(f"progress:{int(idx/total_task*100)}", flush=True)

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