Python：在Engine类内管控子进程的实现合理性及方案探讨

背景 #

我正在实现一个可启动和终止其他进程的程序，已使用subprocess模块创建子进程，通过Popen类启动进程。

from subprocess import Popen, PIPE

class Engine:
   def __init__(self):
    self.processes = list()

   def start_processes(self):
    self.processes.append(Popen(["python","producer.py"], stdout=PIPE, stderr=PIPE))
    self.processes.append(Popen(["python","consumer.py"], stdout=PIPE, stderr=PIPE))

   def end_processes(self):
    for process in self.processes:
        process.kill()

def handle_quit() -> bool:
   u_input = input("Quit?")
   if u_input != "q":
     return False
   else:
     return True

if __name__ == "__main__":
    # Initialise quit flag
    quit_flag = False

    # Create engine
    app_engine = Engine()

    # Start processes
    app_engine.start_processes()

    # Handle input
    while not quit_flag:
        quit_flag = handle_quit()

    # Stop processes
    app_engine.end_processes()

问题 #

我希望能在Engine类内管控所有子进程，请问这种实现方式是否合理？还有其他可行的实现思路吗？

回答 #

当前实现的合理性 #

这种实现基本合理，核心逻辑没问题：用类封装进程的启动、终止逻辑，把进程实例统一存放在列表中管理，符合面向对象的封装原则，能集中管控子进程的生命周期。但存在几个可以优化的细节：

• 用stdout=PIPE和stderr=PIPE但未读取输出，子进程可能因输出缓冲区满而阻塞，无法正常运行；
• kill()是强制终止，可能导致子进程资源未正确释放，优先用terminate()（发送SIGTERM信号让进程优雅退出），无响应时再用kill()；
• 未跟踪子进程的退出状态，无法及时发现进程意外退出并处理；
• start_processes方法硬编码了启动脚本，灵活性不足。

其他可行实现思路 #

思路1：用上下文管理器自动管控生命周期

让Engine类实现__enter__和__exit__方法，借助with语句自动启动、终止进程，避免手动调用时遗漏终止逻辑：

from subprocess import Popen, PIPE, TimeoutExpired

class Engine:
    def __init__(self, process_cmds):
        self.processes = []
        self.process_cmds = process_cmds

    def __enter__(self):
        self.start_processes()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end_processes()

    def start_processes(self):
        for cmd in self.process_cmds:
            self.processes.append(Popen(cmd, stdout=PIPE, stderr=PIPE, text=True))

    def end_processes(self):
        for proc in self.processes:
            proc.terminate()
            try:
                proc.wait(timeout=5)  # 等待5秒让进程优雅退出
            except TimeoutExpired:
                proc.kill()  # 超时则强制终止

# 使用示例
if __name__ == "__main__":
    cmds = [["python", "producer.py"], ["python", "consumer.py"]]
    with Engine(cmds) as engine:
        while input("Quit?") != "q":
            pass

思路2：用第三方库实现精细化管控

如果需要更丰富的进程监控、管理功能（比如查看CPU/内存占用、遍历子进程），可以使用psutil库：

import psutil
from subprocess import Popen

class Engine:
    def __init__(self, process_cmds):
        self.processes = []
        self.process_cmds = process_cmds

    def start_processes(self):
        for cmd in self.process_cmds:
            proc = Popen(cmd)
            self.processes.append(psutil.Process(proc.pid))

    def end_processes(self):
        for proc in self.processes:
            if proc.is_running():
                proc.terminate()
                try:
                    proc.wait(timeout=5)
                except psutil.TimeoutExpired:
                    proc.kill()

思路3：拆分职责，封装单个进程类

把单个子进程的启动、终止、输出处理封装成独立的ManagedProcess类，Engine类仅负责批量管理这些实例，职责更清晰：

from subprocess import Popen, PIPE, TimeoutExpired
import threading

class ManagedProcess:
    def __init__(self, cmd):
        self.cmd = cmd
        self.proc = None
        self._stdout_thread = None
        self._stderr_thread = None

    def _read_output(self, stream, callback):
        # 启动线程读取输出，避免阻塞主进程
        for line in iter(stream.readline, ''):
            callback(line.strip())

    def start(self):
        self.proc = Popen(self.cmd, stdout=PIPE, stderr=PIPE, text=True)
        self._stdout_thread = threading.Thread(target=self._read_output, args=(self.proc.stdout, lambda line: print(f"[{self.cmd}] STDOUT: {line}")))
        self._stderr_thread = threading.Thread(target=self._read_output, args=(self.proc.stderr, lambda line: print(f"[{self.cmd}] STDERR: {line}")))
        self._stdout_thread.start()
        self._stderr_thread.start()

    def stop(self):
        if self.proc and self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=5)
            except TimeoutExpired:
                self.proc.kill()
            # 等待输出线程结束
            self._stdout_thread.join()
            self._stderr_thread.join()

class Engine:
    def __init__(self, process_cmds):
        self.managed_processes = [ManagedProcess(cmd) for cmd in process_cmds]

    def start_processes(self):
        for mp in self.managed_processes:
            mp.start()

    def end_processes(self):
        for mp in self.managed_processes:
            mp.stop()

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