使用asyncio.create_subprocess_shell超时后如何获取部分stdout和stderr

在asyncio环境下，直接用proc.communicate()配合wait_for超时会导致无法获取进程已输出的内容——因为communicate()是一次性读取到进程结束，任务被取消后之前的读取结果会丢失。要解决这个问题，核心思路是手动分流转读取stdout/stderr，而非依赖communicate()的一次性读取逻辑。

实现方案 #

通过创建独立的任务分别读取stdout和stderr流，超时后终止进程并收集已读取的内容：

1. 编写流读取辅助函数

async def read_stream(stream, output_buffer):
    """持续从流中读取数据，存入缓冲区"""
    while True:
        chunk = await stream.read(1024)
        if not chunk:
            break
        output_buffer.append(chunk)

2. 主执行逻辑

async def run_command_with_timeout(cmd, timeout_secs):
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    stdout_buffer = []
    stderr_buffer = []

    # 启动流读取任务
    stdout_task = asyncio.create_task(read_stream(proc.stdout, stdout_buffer))
    stderr_task = asyncio.create_task(read_stream(proc.stderr, stderr_buffer))

    try:
        # 等待进程结束，设置超时
        await asyncio.wait_for(proc.wait(), timeout=timeout_secs)
        # 进程正常结束后，等待读取任务完成
        await asyncio.gather(stdout_task, stderr_task)
    except asyncio.TimeoutError:
        # 超时后先尝试终止进程
        proc.terminate()
        try:
            # 给进程2秒时间优雅退出，否则强制杀死
            await asyncio.wait_for(proc.wait(), timeout=2)
        except asyncio.TimeoutError:
            proc.kill()
        # 等待读取任务完成（进程终止后流会关闭）
        await asyncio.gather(stdout_task, stderr_task)
    finally:
        # 合并缓冲区内容，转为字节串
        stdout = b''.join(stdout_buffer)
        stderr = b''.join(stderr_buffer)
        return stdout, stderr, proc.returncode

方案优势 #

• 即使超时，已读取的输出会被保留在缓冲区中，不会丢失
• 进程终止后仍能读取到其终止前的所有输出
• 逻辑清晰，不依赖communicate()的黑盒行为，可控性更强

为什么原方案不行？ #

原方案中proc.communicate()的任务被wait_for取消后，该任务的内部读取逻辑会中断，且不会返回已读取的部分数据。而手动读取流的方式，读取任务会持续到流关闭（进程终止后流会自动关闭），因此能完整收集到超时前的所有输出。

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