如何线程安全地捕获Python函数调用的stdout输出？

需求与现状 #

现有程序需接入REST API，不愿重写大量代码，计划通过捕获print输出作为接口响应返回。调研相关方案时看到两种思路：

• 明确标注非线程安全的方案：用concurrent.futures.ThreadPoolExecutor测试未出现问题，但认可其存在线程安全隐患；
• 基于线程实现异步的方案：未明确说明自身线程安全性。

自行编写线程安全测试代码（如下），多次测试输出均为['hello world', 'hello world2']，但担忧在FastAPI这类异步REST框架中，concurrent.futures的特性无法适配协程模型，引发输出混乱问题。诉求为寻找异步捕获stdout的可行方案；若不可行，需了解具体原因以便切换替代方案。

测试代码 #

import concurrent.futures
from io import StringIO 
import sys

def main():
    num_tests = 30
    
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = [executor.submit(test) for _ in range(num_tests)]
        
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                print(f"An error occurred: {e}")

class Capturing(list):
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio.getvalue().splitlines())
        del self._stringio    # free up some memory
        sys.stdout = self._stdout
def test():
    with Capturing() as output:
        print('hello world')

    print('displays on screen')

    with Capturing(output) as output:  # note the constructor argument
        print('hello world2')

    print('done')
    print('output:', output)
main()

异步环境下的安全捕获方案 #

FastAPI这类异步框架采用协程模型，多个协程在同一线程内交替执行，修改全局sys.stdout的方案会导致协程间输出互相污染，即使线程池测试通过也无法适配协程场景。推荐以下两种安全方案：

方案1：协程本地存储隔离输出 #

利用Python contextvars模块，为每个异步协程维护独立的stdout缓冲区，避免全局状态冲突：

import contextvars
from io import StringIO
import sys
from fastapi import FastAPI

app = FastAPI()
# 定义协程本地的输出缓冲区
local_stdout = contextvars.ContextVar("local_stdout", default=None)

class AsyncCapturing(StringIO):
    def write(self, s):
        buf = local_stdout.get()
        if buf is not None:
            buf.write(s)
        # 可选：保留原始stdout输出
        sys.__stdout__.write(s)

# 替换全局stdout为适配协程的捕获类
sys.stdout = AsyncCapturing()

def capture_stdout():
    buf = StringIO()
    token = local_stdout.set(buf)
    try:
        yield buf
    finally:
        local_stdout.reset(token)

@app.get("/run")
async def run_program():
    async with capture_stdout() as output:
        # 调用原有输出print的程序逻辑
        print("hello from async endpoint")
        print("another line of output")
    return {"output": output.getvalue().splitlines()}

该方案通过协程本地存储，确保每个请求的输出独立捕获，完全避免协程间干扰。

方案2：子进程隔离执行 #

如果原有程序可独立运行，通过subprocess启动子进程捕获输出，天然隔离请求间的运行环境：

from fastapi import FastAPI
import subprocess

app = FastAPI()

@app.get("/run-program")
async def run_program():
    result = subprocess.run(
        ["python", "your_existing_program.py"],
        capture_output=True,
        text=True
    )
    return {"output": result.stdout.splitlines(), "errors": result.stderr.splitlines()}

优点是无需修改原有代码，完全隔离环境；缺点是进程启动有一定开销，适合非高并发场景。

原有方案不适配异步环境的原因 #

FastAPI基于协程模型，多个协程在同一线程内交替执行，全局sys.stdout是线程共享状态——当协程切换时，未恢复的sys.stdout会导致后续协程的输出被错误捕获，引发数据混乱。而concurrent.futures.ThreadPoolExecutor的测试通过，是因为每个线程有独立的全局状态，线程间修改不会互相影响，但这和协程的执行模型本质不同。

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