异步函数中使用contextlib.redirect_stdout时其他任务输出被误重定向的解决方法

你遇到的问题核心是contextlib.redirect_stdout修改的是全局sys.stdout，而异步任务共享同一线程的全局状态，导致并发任务的输出互相干扰。以下是两种可行的解决思路：

方案一：任务本地存储隔离输出 #

利用Python的contextvars模块实现任务级别的输出隔离，给每个异步任务绑定独立的输出目标，通过代理类动态路由输出：

import asyncio
import contextvars
import sys
from io import StringIO

# 定义任务本地的stdout变量，默认使用原stdout
task_stdout = contextvars.ContextVar("task_stdout", default=sys.stdout)

class TaskLocalStdoutProxy:
    def write(self, data):
        task_stdout.get().write(data)
    
    def flush(self):
        task_stdout.get().flush()

# 替换全局stdout为代理对象
original_stdout = sys.stdout
sys.stdout = TaskLocalStdoutProxy()

async def async_task(task_id, redirect_to_file=False):
    # 确定当前任务的输出目标
    output_target = open("file.txt", "a") if redirect_to_file else StringIO()
    # 将输出目标绑定到当前任务
    token = task_stdout.set(output_target)
    try:
        # 模拟无法控制的第三方输出
        print(f"Task {task_id}: 这是需要隔离的输出")
        await asyncio.sleep(0.1)
    finally:
        # 重置任务本地变量
        task_stdout.reset(token)
        # 处理输出目标
        if redirect_to_file:
            output_target.close()
        else:
            # 将非重定向的输出打印到原控制台
            print(output_target.getvalue(), file=original_stdout, end="")

async def main():
    # 并发执行2个任务，一个重定向到文件，一个正常输出
    await asyncio.gather(
        async_task(1, redirect_to_file=True),
        async_task(2, redirect_to_file=False)
    )
    # 恢复原stdout
    sys.stdout = original_stdout

if __name__ == "__main__":
    asyncio.run(main())

优点：不影响异步并发性能，每个任务的输出完全隔离；缺点：需要替换全局stdout为代理类，代码复杂度稍高。

方案二：异步锁控制全局stdout修改 #

用asyncio.Lock保证同一时间只有一个任务修改全局sys.stdout，避免并发冲突：

import asyncio
import sys
from io import StringIO

# 定义异步锁，保护stdout修改操作
stdout_lock = asyncio.Lock()

async def async_task(task_id, redirect_to_file=False):
    output_target = open("file.txt", "a") if redirect_to_file else StringIO()
    original_stdout = sys.stdout
    # 加锁后修改全局stdout
    async with stdout_lock:
        sys.stdout = output_target
        try:
            print(f"Task {task_id}: 这是需要隔离的输出")
            await asyncio.sleep(0.1)
        finally:
            # 恢复原stdout
            sys.stdout = original_stdout
            # 处理输出目标
            if redirect_to_file:
                output_target.close()
            else:
                print(output_target.getvalue(), end="")

async def main():
    await asyncio.gather(
        async_task(1, redirect_to_file=True),
        async_task(2, redirect_to_file=False)
    )

if __name__ == "__main__":
    asyncio.run(main())

优点：代码逻辑简单，不需要修改全局stdout结构；缺点：加锁会让涉及stdout重定向的任务串行执行，并发量大时会影响性能。

不需要接受当前的混乱输出状态，以上两种方案都能实现你需要的“仅指定任务的输出写入file.txt”的需求。

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