Python实现外部程序实时输出捕获与多端转发的进程通信方案咨询

我来帮你搞定这个问题！你遇到的核心痛点是subprocess默认会缓冲输出，直到进程结束才返回结果，这对无限循环类的长运行程序完全不适用。下面我会给你两种可行的实现方案，以及如何把输出分发到日志、Telnet这类多目标的具体代码。

1. 先给子进程做个小调整（可选但强烈推荐） #

你的第一个脚本first_script.py里，print(i)默认是行缓冲，但如果子进程运行在非交互式环境下（比如被wrapper调用），Python可能会自动改成块缓冲，导致输出没法实时传递出来。所以最好给print加上flush=True，强制刷新缓冲区：

import time

i = 0
running = True
while running:
    print(i, flush=True)  # 关键：强制刷新，让输出立刻被读取
    time.sleep(10)
    if i == 20:
        running = False
    else:
        i += 1
print("Exit reached", flush=True)

2. Wrapper的两种实现方案 #

方案一：事件驱动（回调式，推荐） #

这种方式是实时监听子进程的输出流，一旦有新的行输出，立刻触发处理逻辑（比如写日志、发Telnet），效率更高，实时性也更好，不需要定期轮询。

实现思路：

• 用subprocess.Popen启动子进程，重定向标准输出/错误
• 单独开一个线程读取子进程的输出行，避免阻塞主线程
• 定义统一的输出处理器接口，把日志、Telnet都做成处理器，每读到一行就分发给所有处理器

代码示例：

import subprocess
import threading
import logging
import telnetlib
import time

# 初始化日志记录器，同时输出到文件和控制台
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(message)s',
    handlers=[logging.FileHandler('app.log'), logging.StreamHandler()]
)

# 定义输出处理器的基类，统一接口
class OutputHandler:
    def send(self, line):
        raise NotImplementedError("子类必须实现send方法")

# 日志处理器：把输出写入日志
class LogHandler(OutputHandler):
    def send(self, line):
        logging.info(f"子进程输出: {line.strip()}")

# Telnet处理器：把输出发送到Telnet服务器
class TelnetHandler(OutputHandler):
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.tn = None
        # 初始化时尝试连接Telnet
        self._connect()
    
    def _connect(self):
        try:
            self.tn = telnetlib.Telnet(self.host, self.port)
            logging.info(f"成功连接Telnet服务器 {self.host}:{self.port}")
        except Exception as e:
            logging.error(f"连接Telnet失败: {str(e)}")
    
    def send(self, line):
        # 如果连接断开，尝试重连
        if not self.tn:
            self._connect()
        if self.tn:
            try:
                self.tn.write(f"{line.strip()}\n".encode('utf-8'))
            except Exception as e:
                logging.error(f"发送到Telnet失败: {str(e)}")
                self.tn = None  # 标记连接失效，下次尝试重连

def read_subprocess_output(proc, handlers):
    """读取子进程输出并分发给所有处理器"""
    # 逐行读取输出，直到进程结束
    for line in iter(proc.stdout.readline, ''):
        if not line:
            break
        # 分发给每个处理器
        for handler in handlers:
            handler.send(line)
    # 读取进程退出后剩余的输出
    remaining_output = proc.stdout.read()
    if remaining_output:
        for handler in handlers:
            handler.send(remaining_output)

def run_wrapper():
    # 启动子进程，设置参数确保实时输出
    proc = subprocess.Popen(
        ['python', 'first_script.py'],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,  # 把错误输出也合并到标准输出一起处理
        bufsize=1,  # 行缓冲模式
        text=True  # 直接返回字符串，不需要手动解码字节流
    )

    # 初始化多个输出处理器
    handlers = [
        LogHandler(),
        TelnetHandler('你的Telnet主机', 23)  # 替换成实际的Telnet地址和端口
    ]

    # 启动线程读取输出，主线程可以继续做其他事情
    output_thread = threading.Thread(target=read_subprocess_output, args=(proc, handlers))
    output_thread.start()

    # 等待子进程结束，然后等待输出线程完成
    proc.wait()
    output_thread.join()
    print("子进程已退出")

if __name__ == "__main__":
    run_wrapper()

方案二：主动轮询（适合简单场景） #

如果你更习惯每隔一段时间主动去获取输出，那可以用一个线程把子进程的输出存入缓冲区（比如队列），主线程定期读取缓冲区的内容。

代码示例：

import subprocess
import threading
import logging
import time
from queue import Queue

# 初始化日志记录器
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

def read_subprocess_output(proc, output_queue):
    """读取子进程输出并放入队列（线程安全的缓冲区）"""
    for line in iter(proc.stdout.readline, ''):
        if not line:
            break
        output_queue.put(line)
    # 读取进程退出后的剩余输出
    remaining_output = proc.stdout.read()
    if remaining_output:
        output_queue.put(remaining_output)

def run_wrapper():
    proc = subprocess.Popen(
        ['python', 'first_script.py'],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
        text=True
    )

    # 用队列作为线程安全的输出缓冲区
    output_queue = Queue()
    output_thread = threading.Thread(target=read_subprocess_output, args=(proc, output_queue))
    output_thread.start()

    # 主动轮询：每隔5秒读取一次缓冲区
    while proc.poll() is None:  # 子进程还在运行
        time.sleep(5)
        # 读取队列中所有的输出内容
        output_lines = []
        while not output_queue.empty():
            output_lines.append(output_queue.get())
        if output_lines:
            print("=== 轮询到的输出 ===")
            for line in output_lines:
                line_stripped = line.strip()
                print(line_stripped)
                # 发送到日志
                logging.info(f"子进程输出: {line_stripped}")
                # 发送到Telnet的逻辑可以在这里添加，比如调用TelnetHandler的send方法

    # 子进程结束后，读取剩余的输出
    print("=== 子进程退出，剩余输出 ===")
    while not output_queue.empty():
        line = output_queue.get().strip()
        print(line)
        logging.info(f"子进程输出: {line}")
    
    output_thread.join()
    print("子进程已退出")

if __name__ == "__main__":
    run_wrapper()

3. 关键注意事项 #

• 缓冲区问题：必须确保子进程的输出是行缓冲的，要么子进程的print加flush=True，要么启动子进程时设置bufsize=1和text=True，不然还是会出现输出延迟。
• 线程安全：如果用列表做缓冲区，一定要加threading.Lock保护；用queue.Queue的话，它本身就是线程安全的，不用额外加锁。
• 异常处理：Telnet连接可能会断开，所以处理器里要加自动重连逻辑；子进程可能异常退出，要通过proc.poll()判断进程状态。
• 错误输出处理：示例中把stderr合并到stdout一起处理，如果需要分开处理，可以单独重定向stderr=subprocess.PIPE，再启动一个线程读取stderr。

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