Python原生调用函数时持续重定向stdout至AMQP并获取返回值

场景：有一个const_print函数，执行时会持续向stdout输出状态信息，最终返回Python对象。需要在原生调用（不使用子进程）、不修改原函数的前提下，实时捕获其stdout内容发送至AMQP后端，同时获取函数返回值。此前用子进程实现了捕获输出，但无法拿到返回的Python对象，寻求解决方案。

原函数代码：

import sys
import time

def const_print():
    count = 0
    while True:
        sys.stdout.write(f"Waited for {count} seconds\n")
        sys.stdout.flush()
        count += 1
        time.sleep(1)

        if count > 5:
            sys.stdout.write(f"Waited long enough. Exiting...\n")
            sys.stdout.flush()
            break

    return 'result of intense calculation'

解决方案 #

核心思路是临时替换sys.stdout为自定义输出流，在自定义流的write方法中实时将内容发送至AMQP，同时可保留原函数的控制台输出逻辑，执行完函数后恢复原stdout。这样既能捕获输出，又能直接获取函数返回值。

实现步骤 #

• 自定义继承自io.TextIOBase的输出流类，实现write和flush方法，在write中处理AMQP消息发送
• 调用函数前替换sys.stdout，执行完成后恢复原stdout，避免影响后续代码

完整代码示例 #

import sys
import time
import io
import pika  # 以pika为例作为AMQP客户端，可替换为你使用的AMQP库

# 初始化AMQP连接与通道（根据实际环境配置参数）
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='QUEUE')

class AMQPStdoutRedirector(io.TextIOBase):
    def __init__(self, original_stdout, amqp_channel, routing_key):
        self.original_stdout = original_stdout
        self.amqp_channel = amqp_channel
        self.routing_key = routing_key

    def write(self, data):
        # 实时发送内容到AMQP
        if data.strip():  # 可选：过滤空行
            self.amqp_channel.basic_publish(
                exchange='',
                routing_key=self.routing_key,
                body=data.encode('utf-8')
            )
        # 可选：同步输出到原stdout，保留控制台打印
        self.original_stdout.write(data)
        self.original_stdout.flush()

    def flush(self):
        self.original_stdout.flush()

# 原生调用函数并处理输出
if __name__ == '__main__':
    original_stdout = sys.stdout
    try:
        # 替换stdout为自定义流
        sys.stdout = AMQPStdoutRedirector(original_stdout, channel, 'QUEUE')
        # 直接调用函数，获取返回值
        function_result = const_print()
        print(f"函数返回值：{function_result}")
    finally:
        # 恢复原stdout
        sys.stdout = original_stdout
        # 关闭AMQP连接
        connection.close()

关键细节 #

• 实时性保障：原函数中调用了sys.stdout.flush()，自定义流同步触发原stdout的flush，确保输出内容被实时发送到AMQP
• 兼容性：自定义类遵循Python标准IO流接口，不会干扰原函数的执行逻辑
• 资源安全：通过try...finally块确保无论函数执行是否异常，都会恢复原stdout并关闭AMQP连接，避免资源泄漏

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