如何捕获ProcessPoolExecutor全部输出并重定向至主进程标准输出

ProcessPoolExecutor 默认采用spawn启动流程创建子进程时，每个子进程会启动全新的Python解释器实例，重新导入主模块与所有依赖模块，拥有独立的标准输入输出句柄：

• 主进程中使用contextlib.redirect_stdout做的重定向仅在主进程内生效，无法自动传递给子进程
• 模块顶层的打印、警告输出会在子进程导入模块阶段直接写入子进程自身的stdout/stderr，不会出现在主进程的输出流中
• 直接将主进程的StringIO等内存对象传给子进程无效，进程间地址空间完全隔离，子进程操作的是对象的独立副本

方案1：全量捕获子进程所有输出（含导入阶段、任务执行阶段的print/警告/报错） #

通过进程池的initializer钩子在子进程启动时重定向标准流，搭配多进程管道+主进程守护线程异步回收输出，统一汇入主进程输出流。

第一步：修正模块顶层执行逻辑（可选，针对不需要在导入时执行的代码） #

如果不希望子进程导入模块时触发顶层打印，将func.py中不需要在导入时执行的代码放入__main__判断块：

# func.py
def f(x):
    return x

if __name__ == "__main__":
    # 仅直接运行本文件时执行，被导入时不触发
    print("imported")

第二步：主进程实现输出捕获逻辑 #

import sys
import io
import threading
from multiprocessing import Pipe
from concurrent.futures import ProcessPoolExecutor
from contextlib import redirect_stdout, redirect_stderr
from func import f

# 子进程初始化：重定向标准流到管道
def _worker_init(write_conn):
    class PipeWriter:
        def __init__(self, conn):
            self.conn = conn
        def write(self, content):
            if content:
                self.conn.send(content)
        def flush(self):
            pass
    # 同时重定向标准输出、标准错误，覆盖print、警告、异常栈输出
    sys.stdout = PipeWriter(write_conn)
    sys.stderr = PipeWriter(write_conn)

if __name__ == "__main__":
    # 创建跨进程管道，duplex=False表示单向传输（子进程写，主进程读）
    read_conn, write_conn = Pipe(duplex=False)
    all_output = io.StringIO()

    # 守护线程异步读管道，避免管道缓冲区写满阻塞子进程
    def _collect_output():
        while True:
            try:
                data = read_conn.recv()
                if data is None:
                    break
                # 同步打印到主进程原始标准输出，同时存入缓存
                sys.__stdout__.write(data)
                all_output.write(data)
            except EOFError:
                break
    collect_thread = threading.Thread(target=_collect_output, daemon=True)
    collect_thread.start()

    # 主进程自身的输出重定向正常生效
    with redirect_stdout(all_output), redirect_stderr(all_output):
        # 传入初始化函数和参数，每个子进程启动时自动执行重定向
        with ProcessPoolExecutor(initializer=_worker_init, initargs=(write_conn,)) as ex:
            futures = [ex.submit(f, i) for i in range(15)]
            for fut in futures:
                fut.result()

    # 子进程全部执行完毕后发送终止信号，回收资源
    write_conn.send(None)
    collect_thread.join(timeout=1)
    write_conn.close()
    read_conn.close()

    # 此处all_output.getvalue()包含主进程+所有子进程的全部输出

方案2：仅捕获警告信息（轻量场景） #

如果只需要收集警告不需要捕获全量print，不需要重定向整个标准流，直接替换子进程的警告处理器即可，性能开销更低：

import warnings
from multiprocessing import Queue

# 全局变量存子进程内的队列引用
_worker_q = None

def _custom_warn_handler(message, category, filename, lineno, file=None, line=None):
    warn_content = warnings.formatwarning(message, category, filename, lineno, line)
    _worker_q.put(warn_content)

def _worker_warn_init(q):
    global _worker_q
    _worker_q = q
    # 替换默认警告输出函数
    warnings.showwarning = _custom_warn_handler

# 主进程逻辑和方案1类似，起守护线程读Queue中的警告信息，统一打印/存储即可

• 跨平台场景优先使用multiprocessing.Pipe/multiprocessing.Queue做进程间通信，不要直接用os.pipe，避免Windows下句柄继承异常
• 必须用单独的守护线程异步读取管道/队列，否则子进程输出量超过管道缓冲区大小（通常64KB）时会永久阻塞
• Linux平台可通过multiprocessing.set_start_method("fork")使用fork启动模式，子进程会自动继承主进程的IO重定向配置，无需单独写初始化逻辑，但fork模式在多线程场景下存在死锁风险，不推荐通用场景使用

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