如何在Python多进程中实时捕获函数的stdout与stderr

问题背景 #

我有一个Python函数，希望通过multiprocessing包将其作为独立进程运行，函数代码如下：

def run(ctx: Context):
    print("hello world!")
    return ctx

启动进程的脚本为：

import multiprocessing

p = multiprocessing.Process(target=run, args=(ctx, ))
p.start()
p.join()

需要实时捕获该进程的stdout和stderr，类似subprocess的实现逻辑，但希望直接传入函数而非执行外部命令，该如何实现？

解决方案 #

可以通过以下几种方法实现，无需依赖subprocess.Popen：

方法1：使用multiprocessing.Pipe实时重定向输出

通过管道在父子进程间传递输出，支持实时读取：

import multiprocessing
import sys
from contextlib import redirect_stdout, redirect_stderr

def run(ctx, stdout_pipe, stderr_pipe):
    # 将子进程的stdout和stderr重定向到管道
    with redirect_stdout(stdout_pipe), redirect_stderr(stderr_pipe):
        print("hello world!")
        # 模拟一条stderr输出
        print("this is an error message", file=sys.stderr)
    return ctx

if __name__ == "__main__":
    # 创建双向管道，父进程端用于读取，子进程端用于写入
    parent_stdout, child_stdout = multiprocessing.Pipe()
    parent_stderr, child_stderr = multiprocessing.Pipe()

    ctx = "sample_context"
    p = multiprocessing.Process(target=run, args=(ctx, child_stdout, child_stderr))
    p.start()

    # 父进程关闭子进程端的管道（只保留读端）
    child_stdout.close()
    child_stderr.close()

    # 实时读取stdout内容
    print("=== 捕获的stdout内容 ===")
    while parent_stdout.poll():
        line = parent_stdout.readline()
        if line:
            print(line.decode().strip())

    # 实时读取stderr内容
    print("\n=== 捕获的stderr内容 ===")
    while parent_stderr.poll():
        line = parent_stderr.readline()
        if line:
            print(line.decode().strip())

    p.join()

方法2：使用multiprocessing.Queue收集输出

适合异步收集场景，无需实时轮询，进程结束后统一获取输出：

import multiprocessing
import sys
from contextlib import redirect_stdout, redirect_stderr
import io

def run(ctx, stdout_queue, stderr_queue):
    # 用StringIO临时捕获输出
    stdout_buf = io.StringIO()
    stderr_buf = io.StringIO()
    with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
        print("hello world from child process")
        print("this is a stderr output", file=sys.stderr)
    # 将捕获的内容放入队列
    stdout_queue.put(stdout_buf.getvalue())
    stderr_queue.put(stderr_buf.getvalue())
    return ctx

if __name__ == "__main__":
    stdout_queue = multiprocessing.Queue()
    stderr_queue = multiprocessing.Queue()

    ctx = "test_context"
    p = multiprocessing.Process(target=run, args=(ctx, stdout_queue, stderr_queue))
    p.start()

    # 从队列中取出输出内容
    stdout_content = stdout_queue.get()
    stderr_content = stderr_queue.get()

    print("=== 捕获的stdout内容 ===")
    print(stdout_content)
    print("\n=== 捕获的stderr内容 ===")
    print(stderr_content)

    p.join()

方法3：底层文件描述符重定向

直接替换子进程的stdout/stderr文件描述符，适合需要更底层控制的场景：

import multiprocessing
import sys
import os

def run(ctx):
    print("hello world from child process")
    print("this is stderr output", file=sys.stderr)
    return ctx

if __name__ == "__main__":
    # 创建管道的文件描述符
    r_stdout, w_stdout = os.pipe()
    r_stderr, w_stderr = os.pipe()

    ctx = "my_context"
    p = multiprocessing.Process(target=run, args=(ctx,))
    # 替换子进程的stdout和stderr
    p.stdout = os.fdopen(w_stdout, 'w')
    p.stderr = os.fdopen(w_stderr, 'w')
    p.start()

    # 父进程关闭写端，避免资源泄漏
    os.close(w_stdout)
    os.close(w_stderr)

    # 读取stdout
    print("=== 捕获的stdout内容 ===")
    with os.fdopen(r_stdout, 'r') as f:
        for line in f:
            print(line.strip())

    # 读取stderr
    print("\n=== 捕获的stderr内容 ===")
    with os.fdopen(r_stderr, 'r') as f:
        for line in f:
            print(line.strip())

    p.join()

注意事项 #

• 使用管道时，务必关闭不需要的端点，否则可能导致进程死锁
• 若子进程输出量较大，需确保父进程及时读取，避免管道缓冲区被占满导致阻塞
• redirect_stdout和redirect_stderr是Python 3.4及以上版本的特性，兼容性良好

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