如何将C/C++模块的stderr实时重定向至自定义Python日志处理器？

我需要捕获一个无法修改的C/C++模块输出的std::cerr内容，在输出到终端前进行自定义预处理。目前我用上下文管理器把stderr重定向到临时文件，等上下文结束后再读取处理，但我希望在输出生成时就立即处理，而不是等待上下文结束。虽然可以用线程定时读取临时文件，但想知道有没有更优雅的方案？

当前代码（事后读取临时文件）：

import contextlib
import sys
import os
import tempfile
import io

with redirect_stderr():
    # foo() 执行 `std::cerr << "some log" << std::endl`
    some_module.foo()

@contextlib.contextmanager
def redirect_stderr():
    """Context manager which redirect all stderr output to."""
    # Original stderr file descriptor id
    stderr_fd = sys.stderr.fileno()
    # Duplcate original file descriptor to be able to restore it
    dup_stderr_fd = os.dup(stderr_fd)
    # Create new file descriptor
    tfile = tempfile.TemporaryFile(mode='w+b')

    def _redirect_stream(to_fd):
        # Flush and close original file descriptor
        sys.stderr.close()
        # Use the tempfile instead of stderr
        os.dup2(to_fd, stderr_fd)
        # Tell Python to use the new file to write to stderr
        sys.stderr = os.fdopen(stderr_fd, 'w')

    try:
        # Replace stderr by the new temp file
        _redirect_stream(tfile.fileno())
        yield
    finally:
        # Replace the new temp file by the original stderr
        _redirect_stream(dup_stderr_fd)
        tfile.flush()
        tfile.seek(0, io.SEEK_SET)
        print(tfile.read())

当然有更优雅的实时处理方案！ #

临时文件的问题在于无法实时感知新内容，必须等写入结束后才能读取。我们可以用**管道（os.pipe）**替代临时文件，配合线程监听管道的读端，这样C模块每输出一行内容，就能立刻触发你的自定义处理逻辑，完全不需要等待上下文管理器结束。

方案实现：管道+监听线程 #

import contextlib
import sys
import os
import threading
import io

# 你的自定义日志处理器
class MyCustomLogger:
    @staticmethod
    def write(message):
        # 这里替换成你的自定义预处理逻辑
        print(f"[自定义处理后] {message.strip()}")

@contextlib.contextmanager
def redirect_stderr_realtime():
    # 保存原始stderr的文件描述符
    original_stderr_fd = sys.stderr.fileno()
    dup_stderr_fd = os.dup(original_stderr_fd)  # 备份，用于恢复

    # 创建管道：rfd是读端，wfd是写端
    pipe_read_fd, pipe_write_fd = os.pipe()
    stop_event = threading.Event()  # 用于通知监听线程停止

    def _pipe_listener():
        """持续读取管道内容，实时处理"""
        # 用io包装读端，方便按行读取（兼容文本模式）
        with os.fdopen(pipe_read_fd, 'r', encoding='utf-8') as pipe_reader:
            while not stop_event.is_set():
                # readline()会阻塞，直到有新内容或管道被关闭
                line = pipe_reader.readline()
                if line:
                    # 立刻调用自定义处理器
                    MyCustomLogger.write(line)
                # 管道写端关闭且无内容时，退出循环
                elif stop_event.is_set():
                    break

    def _redirect_stderr(to_fd):
        """将stderr重定向到指定文件描述符"""
        sys.stderr.close()
        os.dup2(to_fd, original_stderr_fd)  # 替换底层文件描述符
        sys.stderr = os.fdopen(original_stderr_fd, 'w')  # 更新Python的sys.stderr引用

    try:
        # 启动监听线程（守护线程，避免程序退出时残留）
        listener_thread = threading.Thread(target=_pipe_listener, daemon=True)
        listener_thread.start()

        # 将stderr重定向到管道的写端
        _redirect_stderr(pipe_write_fd)

        yield  # 执行上下文内的代码

    finally:
        # 停止监听线程
        stop_event.set()
        # 恢复原始stderr
        _redirect_stderr(dup_stderr_fd)
        # 关闭管道写端，让读端收到EOF，触发线程退出
        os.close(pipe_write_fd)
        # 等待监听线程完全结束
        listener_thread.join()

# 使用示例
with redirect_stderr_realtime():
    some_module.foo()

这个方案的优势： #

• 真正实时：管道的readline()是阻塞式的，C模块每输出一行，线程就能立刻处理，不需要定时轮询临时文件
• 资源高效：线程仅在有数据时工作，没有额外的轮询开销
• 可靠性高：通过stop_event和线程join()保证资源正确释放，不会残留僵尸线程
• 兼容性好：直接操作底层文件描述符，完美适配C/C++模块的std::cerr输出

为什么不推荐替换sys.stderr？ #

有些同学会想到直接替换Python的sys.stderr为自定义流，但这个方法对C/C++模块完全无效：因为C模块的std::cerr是直接操作底层的文件描述符（通常是2），绕过了Python的sys.stderr包装，所以替换后C模块的输出还是会直接到终端，不会经过你的自定义处理。

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