如何将Python同步重耗时函数的stdout实时重定向至异步用户推送接口

核心设计前提 #

两个硬约束必须满足：

• 不修改my_heavy_function的原有逻辑，保证其他同步流程复用这个函数时完全不受影响
• 解决同步阻塞函数和异步事件循环的冲突——同步函数运行时会占用当前线程，异步逻辑根本拿不到执行权，没办法主动等待推送方法
• 保证输出的实时性，和终端看到的print打印节奏一致，不做批量攒包

第一步：实现带队列解耦能力的自定义输出流 #

自定义输出流不需要复杂的缓存逻辑，核心是拦截print的写入动作，把收到的内容直接丢进跨线程安全队列，同时保留原有系统stdout的输出能力，不影响命令行调试体验。

import sys
import queue
import threading
import asyncio
from contextlib import contextmanager

# 跨线程安全队列：同步线程写print内容，异步主循环读内容推送
output_queue = queue.Queue()

class StreamingCaptureStdout:
    def __init__(self, original_stdout):
        self.original_stdout = original_stdout

    def write(self, content: str):
        # 原样输出到原有系统stdout，保留终端日志能力
        self.original_stdout.write(content)
        # 跳过空字符串类的无意义内容，有实际打印内容就入队
        if content.strip():
            output_queue.put(content.strip())

    def flush(self):
        # 兼容print的flush调用，无额外逻辑
        self.original_stdout.flush()

第二步：实现输出捕获的上下文管理器 #

用上下文管理器做临时stdout替换，进入上下文时把系统stdout换成自定义捕获流，退出上下文时恢复原有stdout，同时发送队列结束标记，通知异步消费端没有新内容了。

@contextmanager
def capture_heavy_output():
    original_stdout = sys.stdout
    sys.stdout = StreamingCaptureStdout(original_stdout)
    try:
        yield
    finally:
        sys.stdout = original_stdout
        # 入队结束标记，触发异步消费循环退出
        output_queue.put(None)

第三步：跨线程执行同步函数+异步消费推送 #

因为同步函数是阻塞的，必须放到独立后台线程运行，主异步循环不被卡住，持续从队列拉取新打印内容，调用已有的push_message_to_user推送到Web界面。

async def run_heavy_task_with_push():
    # 后台线程执行同步重函数，不阻塞主异步循环
    def run_sync_function():
        with capture_heavy_output():
            my_heavy_function()
    
    heavy_task_thread = threading.Thread(target=run_sync_function, daemon=True)
    heavy_task_thread.start()

    # 实时消费队列里的打印内容
    while True:
        try:
            # 用线程池做非阻塞队列拉取，超时100ms，避免卡住事件循环
            message = await asyncio.get_event_loop().run_in_executor(
                None, lambda: output_queue.get(timeout=0.1)
            )
        except queue.Empty:
            # 队列暂时无内容，切回事件循环处理其他请求
            await asyncio.sleep(0)
            continue

        # 收到结束标记，函数执行完成，退出循环
        if message is None:
            break
        
        # 调用现有异步方法推送到Web界面
        await push_message_to_user(message)

    # 等待后台线程收尾
    heavy_task_thread.join()

关键注意事项 #

• 不依赖StringIO做捕获：StringIO是块式缓存存储，没有实时通知机制，容易出现函数运行完才一次性读出所有内容，达不到实时推送的效果。
• 严禁直接在异步上下文里调用同步重函数：同步函数的阻塞特性会完全卡死异步事件循环，导致整个Web服务无响应。
• 严禁在自定义stdout的write方法里直接调用异步逻辑：write方法运行在同步线程的上下文里，直接await会触发语法错误，完全跑不通。
• 后台线程设置为daemon=True：避免主程序退出时被后台线程阻塞，保证进程能正常回收资源。

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