Python 中安全管理并发进程：启动、监控与协同终止

本文介绍如何在 Python 中可靠地并发启动两个外部程序（如 process1.exe 和 process2.exe），实现进程间状态联动——当 process2 异常退出时立即终止 process1，或在其正常结束后优雅关闭 process1，并确保主流程同步协调不出现 pickle 错误或资源竞争。

在实际自动化脚本开发中，经常需要协同管理多个外部进程——比如仿真器、服务守护进程等。说起来简单，但真正落地时，核心挑战往往集中在几个地方：如何避免进程失控？如何跨进程感知对方的状态？怎么避开序列化陷阱（比如那个经典的 TypeError: cannot pickle 'weakref.ReferenceType'）？以及如何保证资源清理的原子性？ 之前有人尝试用 multiprocessing.Process 把监控逻辑扔到子进程里，结果误把 psutil.Process 实例或 subprocess.Popen 对象直接传了进去——这些对象内部持有不可序列化的 weakref 或系统句柄，在 Windows 下 spawn 启动方式直接就崩了，ForkingPickler.dump() 报错就是典型症状。

✅ 正确实践：以主进程为中心，轻量监控 + 显式生命周期控制

推荐的方案是 “主进程直管 + 简洁事件通信” 模式，不搞花哨的多进程通信，避免跨进程传递复杂对象。几个关键原则：

• 进程启动由主进程完成：直接用 subprocess.Popen 拿到 Popen 实例，然后调用 .wait() 或轮询 .poll()，干净利落。
• 状态检查逻辑保留在主进程内：没必要单独开一个子进程去跑 checkStatus——除非那个检查本身耗时极长且真的需要并行（这种情况下也应该重构为纯数据/信号驱动）。
• 终止逻辑集中决策：process2 一结束，立马触发 process1.terminate()，不需要额外守护进程来盯着。
• 异常退出兜底：用 try/except 配合 process1.poll() is not None 判断是否已经终止，防止重复 kill。

下面是一个优化后的完整可运行示例，兼容 Windows、macOS 和 Linux：

import subprocess
import time
import sys

def check_status():
    """模拟检查逻辑：可在此处加入日志、API 调用、文件校验等"""
    print("[INFO] Starting pre-checks...")
    time.sleep(3)  # 模拟耗时操作
    print("[INFO] Checks completed successfully.")

def main():
    # 启动两个进程（注意路径和权限）
    try:
        process1 = subprocess.Popen(
            ["./process1.exe"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
        )
        process2 = subprocess.Popen(
            ["./process2.exe"],
            creationflags=subprocess.CREATE_NEW_CONSOLE if sys.platform == "win32" else 0
        )
    except FileNotFoundError as e:
        print(f"[ERROR] Failed to launch executable: {e}")
        return

    print(f"[INFO] Launched process1 (PID: {process1.pid}), process2 (PID: {process2.pid})")

    # 执行必要检查（同步阻塞，确保前置条件满足）
    check_status()

    # 主动等待 process2 结束（正常或异常退出均会返回）
    print("[INFO] Waiting for process2 to exit...")
    try:
        process2.wait(timeout=300)  # 最多等待 5 分钟
    except subprocess.TimeoutExpired:
        print("[WARN] process2 timed out; forcing termination.")
        process2.terminate()
        process2.wait()

    # process2 已退出 → 立即终止 process1
    if process1.poll() is None:  # 确保 process1 仍在运行
        print("[INFO] Terminating process1...")
        process1.terminate()
        try:
            process1.wait(timeout=10)
        except subprocess.TimeoutExpired:
            print("[WARN] process1 did not terminate gracefully; killing...")
            process1.kill()
    else:
        print("[INFO] process1 already exited.")

    print("[INFO] Script finished.")

if __name__ == "__main__":
    main()

⚠️ 关键注意事项

• 路径与可执行性：确保 process1.exe 和 process2.exe 在当前工作目录或 PATH 中；建议使用 os.path.abspath() 显式指定路径，避免意外。
• Windows 控制台行为：CREATE_NEW_CONSOLE 仅在 Windows 下有效；Linux/macOS 可以忽略，或者改用 start_new_session=True 达到类似效果。
• 资源释放：始终调用 .wait() 或 .poll() 来避免僵尸进程；.terminate() 之后建议跟上 .wait() 确保回收。
• PyInstaller 兼容性：打包后路径可能发生变化，务必使用 sys._MEIPASS 动态定位资源（例如 os.path.join(sys._MEIPASS, "process1.exe")）。
• 健壮性增强（可选）：如果需要监听 process2 异常退出（比如崩溃），可以用 psutil 在主循环中轮询 PID 是否存在（注意不要把这个对象传入子进程）。示例：

import psutil
while process2.poll() is None:
    if not psutil.pid_exists(process2.pid):
        print("[ALERT] process2 crashed unexpectedly!")
        process1.terminate()
        break
    time.sleep(1)

这个方案放弃了容易出错的多进程通信模型，转而利用 subprocess 原生 API 提供的可靠进程控制能力。逻辑清晰，调试友好，部署稳定，是生产环境中管理外部进程协作的推荐范式。