Python subprocess如何同时设置超时并实时捕获标准输出

可以同时满足「子进程超时控制」和「实时读取标准输出」两个需求，不需要依赖Popen.communicate()，核心逻辑是自行实现超时判断，边逐行读取子进程输出边计时，超时后主动终止子进程即可。

基础版（适合子进程持续输出日志的场景） #

这个版本逻辑简单，和原有代码的适配成本最低：

import subprocess
import time

# 自定义配置
TIMEOUT = 30  # 超时阈值，单位秒
external_script = ["python", "your_external_script.py"]  # 外部脚本启动命令，用列表格式传参

proc = subprocess.Popen(
    external_script,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,  # 自动将字节输出解码为字符串，Python3.7以下版本替换为 universal_newlines=True
    bufsize=1   # 启用行缓冲，保证按行返回输出
)

start_time = time.time()
try:
    for line in proc.stdout:
        # 每次读行前判断是否超时
        if time.time() - start_time > TIMEOUT:
            raise TimeoutError(f"子进程运行时长超过{TIMEOUT}秒")
        # 实时打印日志，去掉行尾自带的换行避免日志空行
        logger.info(line.rstrip("\n"))
    # 等待进程正常结束，获取返回码
    proc.wait()
except TimeoutError:
    # 先发送温和终止信号
    proc.terminate()
    try:
        # 最多等待5秒让子进程自行退出
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        # 子进程未响应终止信号，强制杀死
        proc.kill()
        proc.wait()
    logger.error(f"外部脚本运行超时，已强制终止，超时阈值：{TIMEOUT}秒")
finally:
    # 关闭管道释放资源
    if proc.stdout:
        proc.stdout.close()

增强版（兼容子进程长时间无输出的场景） #

基础版在子进程长时间不打印日志时，会卡在管道读IO的位置，无法及时触发超时。如果外部脚本可能出现长时间无输出的情况，可以用独立守护线程做超时监控，主线程只负责读取日志：

import subprocess
import time
import threading

# 自定义配置
TIMEOUT = 30
external_script = ["python", "-u", "your_external_script.py"]

proc = subprocess.Popen(
    external_script,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1
)

timeout_triggered = False
def timeout_monitor(proc, timeout):
    """独立线程轮询子进程状态，超时后终止进程"""
    global timeout_triggered
    start_ts = time.time()
    # 轮询检查进程状态，直到进程退出
    while proc.poll() is None:
        if time.time() - start_ts > timeout:
            timeout_triggered = True
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait()
            return
        time.sleep(0.5)  # 每0.5秒检查一次，平衡精度和性能开销

# 启动守护线程做超时监控
monitor_thread = threading.Thread(target=timeout_monitor, args=(proc, TIMEOUT), daemon=True)
monitor_thread.start()

# 主线程实时读取输出
for line in proc.stdout:
    if not timeout_triggered:
        logger.info(line.rstrip("\n"))

proc.wait()
if timeout_triggered:
    logger.error(f"外部脚本运行超时，已强制终止，超时阈值：{TIMEOUT}秒")

关键注意点 #

• 启动子进程时优先用列表格式传入命令，尽量不要开启shell=True，避免命令注入风险，也能减少Shell层带来的输出缓冲问题。
• 如果被调用的是Python脚本，非终端环境下Python默认开启块缓冲，会攒够一定量的输出才写入管道，导致日志无法实时打印，启动时可以给Python命令加-u参数禁用缓冲，即命令写为["python", "-u", "target_script.py"]。其他语言编写的程序如果有类似缓冲问题，也建议对应关闭输出缓冲。
• 终止子进程时优先调用terminate()发送常规终止信号，给子进程留出清理资源的时间，若等待后仍未退出再调用kill()发送强制终止信号，终止后必须调用wait()回收进程资源，避免产生僵尸进程。

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