Python中如何仅抑制指定协程的stdout输出且不影响其他协程

你遇到的问题其实是因为contextlib.redirect_stdout修改的是全局的sys.stdout，而Python协程是在同一个线程中切换运行的——当my_coroutine进入重定向上下文后执行到await asyncio.sleep(1)挂起时，sys.stdout仍然保持着重定向到/dev/null的状态，此时事件循环切换到my_coroutine2运行，它的print输出也会被吞进黑洞里，所以你看不到它的内容。

下面给你两种实用的解决方案：

方案一：用线程隔离重定向（适合同步代码块） #

如果需要抑制输出的部分是同步代码（没有await），可以把这部分代码放到线程池里执行，线程级的stdout重定向不会影响主线程的其他协程：

import asyncio
import contextlib
import os
import sys

# 把需要抑制输出的同步代码单独抽出来
def sync_suppressed_block():
    with open(os.devnull, 'w') as f, contextlib.redirect_stdout(f):
        print("This will not be printed")
        # 这里是同步的耗时操作，比如time.sleep
        import time
        time.sleep(1)

async def my_coroutine():
    loop = asyncio.get_running_loop()
    # 在线程池里执行同步代码，隔离stdout重定向
    await loop.run_in_executor(None, sync_suppressed_block)
    print("This will be printed")

async def my_coroutine2():
    print("This will be printed from 2")
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(
        my_coroutine(),
        my_coroutine2()
    )

asyncio.run(main())

方案二：用上下文变量实现异步安全的print（适合异步代码块） #

如果需要抑制输出的部分包含异步操作（有await），可以用contextvars实现任务级的输出隔离，让每个协程的print输出到自己指定的目标：

import asyncio
import os
import sys
import builtins
from contextvars import ContextVar

# 定义上下文变量，保存当前任务的输出目标，默认是系统stdout
print_target: ContextVar[object] = ContextVar("print_target", default=sys.stdout)

# 自定义print函数，使用当前任务的输出目标
def task_safe_print(*args, **kwargs):
    target = print_target.get()
    print(*args, file=target, **kwargs)

# 猴子补丁替换全局的print函数
builtins.print = task_safe_print

async def my_coroutine():
    redirect_token = None
    try:
        with open(os.devnull, 'w') as f:
            # 设置当前任务的输出目标为/dev/null
            redirect_token = print_target.set(f)
            print("This will not be printed")
            await asyncio.sleep(1)
    finally:
        # 恢复当前任务的输出目标
        if redirect_token is not None:
            print_target.reset(redirect_token)
    print("This will be printed")

async def my_coroutine2():
    print("This will be printed from 2")
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(
        my_coroutine(),
        my_coroutine2()
    )

asyncio.run(main())

运行任意一种方案的代码，你都会得到想要的输出：

This will be printed from 2
This will be printed

备注：内容来源于stack exchange，提问作者BAdhi