如何使用asyncio并发执行函数并替代线程处理子进程的stdout与stderr？

我当前的实现逻辑是启动一个子进程，需要同时读取其stdout和stderr。因此在调用子进程后，我为stdout启动新线程，主线程负责处理stderr，相关代码如下：

# imports
from subprocess import Popen, PIPE
from threading import Thread
def handle_stdout(stdout):
    # ... do something with stdout,
    # not relevant to the question
    pass
def my_fn():
    proc = Popen([...], stdout=PIPE, stderr=PIPE)
    Thread(target=lambda: handle_stdout(proc.stdout)).start()
    # ... handle stderr
    print(proc.stderr.read())
    proc.wait()
    proc.kill()
my_fn()

请问是否可以使用asyncio实现相同的功能？

当然可以！asyncio 提供了原生的异步子进程支持，完全可以替代线程方案来同时读取子进程的 stdout 和 stderr，而且代码会更贴合异步编程模型，不需要手动管理线程。

下面是对应的 asyncio 实现版本：

import asyncio

async def handle_stdout(stdout):
    # 异步逐行读取stdout并处理，你可以替换成自己的业务逻辑
    while True:
        data = await stdout.readline()
        if not data:
            break
        # 示例：解码后打印stdout内容
        print(f"STDOUT: {data.decode().strip()}")

async def my_async_fn():
    # 启动异步子进程，替换[...]为你的实际命令，比如["ls", "-l"]
    proc = await asyncio.create_subprocess_exec(
        "...",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    # 创建处理stdout的异步任务，让它和stderr读取并发执行
    stdout_task = asyncio.create_task(handle_stdout(proc.stdout))
    # 异步读取stderr的全部内容（也可以改成逐行处理）
    stderr_data = await proc.stderr.read()
    
    # 等待stdout的处理任务完成，确保所有输出都被处理
    await stdout_task
    
    # 等待子进程正常结束
    await proc.wait()
    
    # 处理stderr内容，示例为打印
    print(f"STDERR: {stderr_data.decode().strip()}")

# 启动异步主函数
asyncio.run(my_async_fn())

关键细节说明： #

• 用 asyncio.create_subprocess_exec 替代传统的 subprocess.Popen，这是asyncio专为异步场景设计的子进程创建API，返回的Process对象支持异步操作。
• asyncio.create_task 会将处理stdout的协程加入事件循环，让它和stderr的读取操作并行执行，无需手动创建线程。
• 读取stdout时用await stdout.readline()实现异步非阻塞读取，避免阻塞整个事件循环；如果不需要逐行处理，也可以用await stdout.read()一次性读取全部内容。
• 必须等待stdout的处理任务完成后再结束子进程，确保所有输出都被处理完毕。

如果你的handle_stdout逻辑不需要逐行处理，也可以简化成一次性读取：

async def handle_stdout(stdout):
    data = await stdout.read()
    # ... 这里处理整个stdout的内容
    print(f"STDOUT full content: {data.decode().strip()}")

这种异步方案相比线程方案的优势是：没有线程切换的开销，所有操作都在同一个事件循环中处理，资源占用更低，代码结构也更统一，特别适合已经采用asyncio架构的项目。

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