基于Flet的异步应用中如何将STDOUT重定向至内置控制台

核心思路 #

sys.stdout的write方法是同步的，无法直接在里面await异步的page.update_async()，因此需要通过异步队列做中间缓冲：

• 自定义继承自io.TextIOBase的流类，将输出内容存入异步队列
• 启动独立异步任务，持续从队列读取内容、更新Text组件并调用update_async()
• 替换sys.stdout和sys.stderr为自定义流

完整代码实现 #

import asyncio
import io
import sys
import flet as ft

class AsyncConsoleStream(io.TextIOBase):
    def __init__(self, queue: asyncio.Queue):
        self.queue = queue

    def write(self, s):
        if s.strip():  # 可选：过滤空行
            asyncio.create_task(self.queue.put(s))
        return len(s)

async def console_consumer(queue: asyncio.Queue, text_area: ft.Text):
    while True:
        content = await queue.get()
        text_area.value += content
        await text_area.page.update_async()
        queue.task_done()

async def main(page: ft.Page):
    page.title = "异步控制台示例"
    console_text = ft.Text(value="", selectable=True, expand=True)
    page.add(console_text)

    # 初始化异步队列与自定义流
    output_queue = asyncio.Queue()
    sys.stdout = AsyncConsoleStream(output_queue)
    sys.stderr = AsyncConsoleStream(output_queue)

    # 启动输出消费任务
    asyncio.create_task(console_consumer(output_queue, console_text))

    # 测试输出
    print("这是标准输出测试内容")
    try:
        1 / 0
    except ZeroDivisionError:
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    ft.app(target=main, view=ft.WEB_BROWSER)

关键细节说明 #

• AsyncConsoleStream的write方法用asyncio.create_task将内容存入队列，避免阻塞同步调用流程
• console_consumer作为常驻异步任务，在异步上下文里安全调用update_async()，完全适配Flet的异步机制
• 替换sys.stdout/stderr后，所有print输出、异常栈信息都会自动流向Text组件

注意事项 #

• 应用关闭时可恢复原始stdout/stderr，避免影响后续代码执行
• 若需自动滚动到最新内容，可在update_async()前添加page.scroll_to_end()
• 大量输出场景下，建议限制Text组件的内容长度，防止内存占用过高

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