如何在 Python 中安全协调并监控多个外部进程的并发执行

本文介绍一种简洁可靠的方案，使用 subprocess 结合 multiprocessing 实现对两个外部可执行文件（如 process1.exe 和 process2.exe）的协同管控：确保 process2 正常退出后才终止 process1，并在 process2 异常崩溃时及时中止 process1 与整个脚本。

本文介绍一种简洁可靠的方案，使用 subprocess 结合 multiprocessing 实现对两个外部可执行文件（如 process1.exe 和 process2.exe）的协同管控：确保 process2 正常退出后才终止 process1，并在 process2 异常崩溃时及时中止 process1 与整个脚本。

在实际自动化或系统集成场景中，常需同时启动并协调多个外部进程（如仿真器、服务代理、CLI 工具等），其生命周期存在强依赖关系：例如 process2 是主控进程，process1 是辅助服务；一旦 process2 意外退出，必须立即终止 process1 并停止后续逻辑，避免资源泄漏或状态不一致。

原代码尝试通过 psutil 轮询 PID 并用多进程监听进程存活状态，但存在多个关键缺陷：

• ❌ 跨进程对象不可序列化：将 multiprocessing.Process 实例（含 weakref 等非可 pickle 对象）作为参数传入子进程，触发 TypeError: cannot pickle ‘weakref.ReferenceType’ object；
• ❌ 竞态与时机错位：findPID() 在进程刚启动时可能因调度延迟未能捕获 PID，导致监控失效；
• ❌ 逻辑耦合过重：killSimunteneously 试图主动 kill 进程并干预 checkStatus 生命周期，违背职责分离原则，且易引发死锁或资源竞争。

✅ 推荐采用更健壮、符合 Python 最佳实践的方案：直接利用 subprocess.Popen.wait() 的阻塞特性 + 显式信号协调，避免轮询与 PID 查找，大幅简化逻辑并提升可靠性。

以下是优化后的完整实现：

立即学习“Python免费学习笔记（深入）”；

import subprocess
import multiprocessing
import sys

def checkStatus(event, queue):
    # ✅ 示例：执行关键校验逻辑（如端口检测、日志就绪判断、API 健康检查等）
    try:
        # 模拟耗时检查（替换为真实业务逻辑）
        import time
        time.sleep(3)
        print("[INFO] Pre-exit checks completed successfully.")
        queue.put("checkStatus Completed!")
    except Exception as e:
        print(f"[ERROR] Check failed: {e}")
        queue.put(f"checkStatus Failed: {e}")
    finally:
        # ✅ 仅当 process2 尚未退出时，才等待其完成（否则 event 已被 set）
        if not event.is_set():
            print("[INFO] Waiting for process2 to exit before proceeding...")
            event.wait()
        print("[INFO] checkStatus exiting.")

def main():
    # ✅ 启动两个进程，保留 process2 的控制台（可选）
    try:
        process1 = subprocess.Popen(["./process1.exe"])
        process2 = subprocess.Popen(
            ["./process2.exe"],
            creationflags=subprocess.CREATE_NEW_CONSOLE
        )
    except FileNotFoundError as e:
        print(f"[FATAL] Failed to launch executable: {e}")
        sys.exit(1)

    queue = multiprocessing.Queue()
    event = multiprocessing.Event()

    # ✅ 启动检查任务（非阻塞）
    p1 = multiprocessing.Process(target=checkStatus, args=(event, queue))
    p1.start()

    # ✅ 核心逻辑：同步等待 process2 退出（天然支持异常终止检测）
    try:
        process2.wait()  # ⚠️ 若 process2 崩溃，此调用立即返回，返回码非零
        print(f"[INFO] process2 exited with code {process2.returncode}")
    except KeyboardInterrupt:
        print("\n[INTERRUPT] User stopped script.")
        process2.terminate()
        process1.terminate()
        sys.exit(1)

    # ✅ process2 已退出 → 安全终止 process1
    process1.terminate()
    try:
        process1.wait(timeout=5)  # 等待优雅退出，超时则强制 kill
    except subprocess.TimeoutExpired:
        process1.kill()
        print("[WARN] process1 did not terminate gracefully; forced kill.")

    # ✅ 通知检查进程结束，并等待其退出
    event.set()
    p1.join(timeout=10)
    if p1.is_alive():
        p1.terminate()
        p1.join()
        print("[WARN] checkStatus process did not exit in time; terminated.")

    # ✅ 输出结果
    while not queue.empty():
        print(queue.get())

if __name__ == "__main__":
    main()

关键改进说明：

• 零轮询、零 PID 查找：直接复用 subprocess.Popen 对象的 wait() 方法，底层由操作系统通知进程状态变更，无延迟、无竞态、无资源浪费；
• 天然支持崩溃检测：process2.wait() 在 process2 异常退出（如 segfault、未处理异常）时立即返回，returncode 为非零值，可据此做差异化处理（如记录错误日志）；
• 职责清晰、解耦可靠：checkStatus 专注业务检查，主流程专注进程生命周期管理；通过 Event 协调而非跨进程操作对方状态；
• 健壮性增强：添加超时机制（wait(timeout=…)）、异常捕获、强制清理路径，确保脚本在各种异常下均能释放资源；
• 兼容 PyInstaller 打包：不依赖 psutil.process_iter() 或动态进程名匹配，避免打包后因路径/权限问题导致的 PID 查找失败。

? 提示：若需在 process2 退出前实时响应其崩溃（而非等待 wait() 返回），可改用 subprocess.Popen.poll() 配合短间隔轮询（如 time.sleep(0.5)），但需权衡 CPU 开销；绝大多数场景下 wait() 已足够高效且语义明确。

该方案兼顾简洁性、可维护性与生产环境鲁棒性，是协调外部进程依赖关系的推荐实践。