Python 3.6调用外部进程：非阻塞轮询结果与超时终止实现

针对你遇到的需求——在Python 3.6里启动一个不会自行终止、输出时机不确定的外部进程，非阻塞轮询其输出，超时后终止进程——这里有两种靠谱的实现方案：

方案一：使用select模块实现非阻塞检查 #

select模块可以帮我们非阻塞地检测文件描述符是否可读，完美解决io.BufferedReader阻塞读取的问题。直接替换你原来的循环逻辑即可：

import subprocess, time, select

# 启动外部进程，universal_newlines=True让输出以字符串返回（Python3.6用这个，3.7+可用text=True）
myprocess = subprocess.Popen(
    ['python', 'mytest.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    universal_newlines=True
)
found_result = False
timeout_seconds = 15
start_time = time.time()

while time.time() - start_time < timeout_seconds:
    # 非阻塞检查stdout是否有可读数据，超时0.1秒避免空转耗CPU
    ready, _, _ = select.select([myprocess.stdout], [], [], 0.1)
    if ready:
        # 读取一行输出（因为select确认有数据，这里不会阻塞）
        line = myprocess.stdout.readline()
        if not line:
            # 进程意外终止的情况
            break
        print(f"收到输出: {line.strip()}")
        # 检查目标结果是否出现
        if 'result=' in line:
            print("找到目标结果！")
            found_result = True
            break
    time.sleep(0.1)

# 确保进程被终止
if myprocess.poll() is None:
    myprocess.kill()
    myprocess.wait()  # 等待进程彻底结束

print(f"进程已终止，是否找到结果: {found_result}")

关键细节： #

• select.select([myprocess.stdout], [], [], 0.1)：第一个参数是要检测的可读文件对象，最后一个参数是超时时间（0.1秒），调用后立即返回，不会阻塞
• universal_newlines=True：将进程的字节流输出转为字符串，省去手动解码的麻烦
• 每次循环检查超时时间，确保不会超过设定的15秒

方案二：用线程后台读取输出，主线程控制超时 #

如果需要更灵活的输出处理（比如同时处理多个进程输出），可以用线程来后台读取，主线程只负责超时控制：

import subprocess, time, threading

def read_process_output(process, output_handler):
    # 循环读取输出，直到进程结束
    for line in iter(process.stdout.readline, ''):
        output_handler(line.strip())

# 启动外部进程
myprocess = subprocess.Popen(
    ['python', 'mytest.py'],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    universal_newlines=True
)
found_result = False

# 定义输出处理函数
def handle_line(line):
    global found_result
    print(f"收到输出: {line}")
    if 'result=' in line:
        found_result = True

# 启动守护线程读取输出（主线程退出时自动终止）
output_thread = threading.Thread(
    target=read_process_output,
    args=(myprocess, handle_line)
)
output_thread.daemon = True
output_thread.start()

# 等待超时或找到结果
timeout_seconds = 15
start_time = time.time()
while time.time() - start_time < timeout_seconds and not found_result:
    time.sleep(0.1)

# 终止进程
if myprocess.poll() is None:
    myprocess.kill()
    myprocess.wait()

print(f"进程已终止，是否找到结果: {found_result}")

两种方案对比： #

• select方案：轻量，无需额外线程，适合简单的单进程轮询场景
• 线程方案：更灵活，能同时处理多个进程的输出，适合复杂场景

为什么你之前尝试的方法不适用？ #

• myprocess.communicate()：会一直等待进程终止，但你的目标进程不会自行结束，所以会卡住
• 先终止再读取：无法提前捕获输出时机，可能错过结果
• readline()：默认是阻塞调用，会一直等到有输出才返回，不符合非阻塞需求

内容的提问来源于stack exchange，提问作者Csaba Faragó