如何在运行时捕获线程中exec()调用的stdout与stderr

你的代码失效的核心原因是：contextlib.redirect_stdout和redirect_stderr是线程局部生效的——主线程中设置的重定向上下文，不会被新创建的线程继承。因此exec的输出依然会直接打印到默认的stdout/stderr，不会写入你定义的StringIO对象。

下面提供两种无需子进程的解决方案：

方案一：用线程安全队列传递输出 #

通过自定义流类将输出写入线程安全队列，主线程轮询队列实时获取内容：

import threading
import time
from io import StringIO
import contextlib
import queue

class QueueStream(StringIO):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def write(self, buf):
        super().write(buf)
        self.queue.put(buf)

def exec_with_capture(code, out_queue, err_queue):
    # 在目标线程内部设置输出重定向
    f = QueueStream(out_queue)
    e = QueueStream(err_queue)
    with contextlib.redirect_stdout(f), contextlib.redirect_stderr(e):
        exec(code, globals(), locals())

# 待执行的代码
code = "import time; print('hello'); time.sleep(2); print('working'); time.sleep(2); print('coming to an end'); time.sleep(2); print('end')"

# 创建线程安全队列用于传递输出
out_queue = queue.Queue()
err_queue = queue.Queue()

# 启动执行线程
t1 = threading.Thread(target=exec_with_capture, args=(code, out_queue, err_queue))
t1.start()

# 实时读取输出
while t1.is_alive():
    # 取出队列中所有可用的stdout内容
    while not out_queue.empty():
        print("捕获到stdout:", out_queue.get_nowait(), end='')
    # 取出队列中所有可用的stderr内容
    while not err_queue.empty():
        print("捕获到stderr:", err_queue.get_nowait(), end='')
    time.sleep(0.5)  # 控制轮询频率，避免占用过多资源

# 读取线程结束后剩余的输出
while not out_queue.empty():
    print("捕获到stdout:", out_queue.get_nowait(), end='')
while not err_queue.empty():
    print("捕获到stderr:", err_queue.get_nowait(), end='')

t1.join()

方案二：用带锁的StringIO缓冲区 #

给StringIO加线程锁保证多线程读写安全，主线程轮询对比内容差异，只打印新增输出：

import threading
import time
from io import StringIO
import contextlib

class LockedStringIO(StringIO):
    def __init__(self):
        super().__init__()
        self.lock = threading.Lock()

    def write(self, buf):
        with self.lock:
            super().write(buf)

    def getvalue(self):
        with self.lock:
            return super().getvalue()

def exec_with_capture(code, out_buf, err_buf):
    # 在目标线程内部设置输出重定向
    with contextlib.redirect_stdout(out_buf), contextlib.redirect_stderr(err_buf):
        exec(code, globals(), locals())

# 待执行的代码
code = "import time; print('hello'); time.sleep(2); print('working'); time.sleep(2); print('coming to an end'); time.sleep(2); print('end')"

# 创建带锁的输出缓冲区
out_buf = LockedStringIO()
err_buf = LockedStringIO()

# 启动执行线程
t1 = threading.Thread(target=exec_with_capture, args=(code, out_buf, err_buf))
t1.start()

last_out = ""
last_err = ""
# 实时读取新增输出
while t1.is_alive():
    current_out = out_buf.getvalue()
    current_err = err_buf.getvalue()
    # 只打印新增的stdout内容
    if current_out != last_out:
        print("新增stdout:", current_out[len(last_out):], end='')
        last_out = current_out
    # 只打印新增的stderr内容
    if current_err != last_err:
        print("新增stderr:", current_err[len(last_err):], end='')
        last_err = current_err
    time.sleep(0.5)

# 打印线程结束后剩余的内容
current_out = out_buf.getvalue()
current_err = err_buf.getvalue()
if current_out != last_out:
    print("新增stdout:", current_out[len(last_out):], end='')
if current_err != last_err:
    print("新增stderr:", current_err[len(last_err):], end='')

t1.join()

关键说明 #

• 必须在目标线程内部设置contextlib.redirect_stdout/redirect_stderr，才能捕获该线程中exec的输出。
• 多线程环境下直接使用原生StringIO会有线程安全问题，要么用线程安全的队列，要么给缓冲区加锁。

内容的提问来源于stack exchange，提问作者Francisco Gama T. R.