Python subprocess stdout.read遇输入提示符永久阻塞问题求助
问题原因分析
你遇到的isatty()和writable()无效的问题,本质是因为:
- 子进程的
stdout被重定向到了管道(PIPE),而非终端设备,所以proc.stdout.isatty()永远返回False。 proc.stdout是父进程用于读取子进程输出的管道端,对父进程来说是只读的,因此proc.stdout.writable()必然返回False。这两个方法根本不适合用来判断子进程是否进入等待输入的状态。
解决方案
下面提供几种可行的处理方式:
方案1:匹配输出提示符文本
最直接的方式是读取子进程输出,当检测到目标提示符(比如And you what is your name:)时,停止读取并发送输入。
修改后的main1.py代码:
from subprocess import Popen, PIPE import sys proc = Popen(["python3", "main2.py"], stdin=PIPE, stdout=PIPE, stderr=PIPE, text=True) def read_until_prompt(): prompt = "And you what is your name:" collected_output = [] while proc.poll() is None: # 逐字符读取,避免整行读取阻塞 char = proc.stdout.read(1) if not char: break sys.stdout.write(char) sys.stdout.flush() collected_output.append(char) # 检查是否匹配提示符 if ''.join(collected_output[-len(prompt):]) == prompt: break read_until_prompt() proc.stdin.write("My name is Catwoman\n") proc.stdin.flush() # 读取子进程剩余输出 remaining = proc.stdout.read() sys.stdout.write(remaining) sys.stdout.flush()
方案2:使用非阻塞IO
通过将子进程的stdout设置为非阻塞模式,避免read()永久阻塞,同时循环检查输出内容。
Linux/macOS 版本
from subprocess import Popen, PIPE import sys import fcntl import os proc = Popen(["python3", "main2.py"], stdin=PIPE, stdout=PIPE, stderr=PIPE, text=True) # 将stdout设置为非阻塞 fd = proc.stdout.fileno() fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) prompt = "And you what is your name:" collected_output = [] while proc.poll() is None: try: output = proc.stdout.read() if output: sys.stdout.write(output) sys.stdout.flush() collected_output.append(output) if prompt in ''.join(collected_output): break except BlockingIOError: # 没有可用输出时短暂休眠,避免CPU空转 import time time.sleep(0.1) proc.stdin.write("My name is Catwoman\n") proc.stdin.flush() # 恢复阻塞模式读取剩余输出 fcntl.fcntl(fd, fcntl.F_SETFL, fl) remaining = proc.stdout.read() sys.stdout.write(remaining) sys.stdout.flush()
Windows 版本
from subprocess import Popen, PIPE import sys import msvcrt proc = Popen(["python", "main2.py"], stdin=PIPE, stdout=PIPE, stderr=PIPE, text=True) prompt = "And you what is your name:" collected_output = [] while proc.poll() is None: # 检查是否有可读取的输出 if msvcrt.kbhit(proc.stdout.fileno()): output = proc.stdout.read() sys.stdout.write(output) sys.stdout.flush() collected_output.append(output) if prompt in ''.join(collected_output): break else: # 短暂休眠 import time time.sleep(0.1) proc.stdin.write("My name is Catwoman\n") proc.stdin.flush() remaining = proc.stdout.read() sys.stdout.write(remaining) sys.stdout.flush()
方案3:使用pexpect库(推荐)
pexpect是专门用于处理交互式子进程的库,能自动等待指定的提示符并发送输入,非常适合这类场景。
首先安装库:
pip install pexpect
修改后的main1.py代码:
import pexpect proc = pexpect.spawn("python3 main2.py", encoding='utf-8') # 等待提示符出现 proc.expect("And you what is your name:") # 打印之前的输出 print(proc.before, end='') print(proc.match.group(), end='') # 发送输入 proc.sendline("My name is Catwoman") # 读取剩余输出并打印 print(proc.read(), end='')
这个方案代码更简洁,处理交互式场景更可靠。
内容的提问来源于stack exchange,提问作者Thomas Aumaitre
相关产品推荐
相关产品推荐

