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

如何在 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版

图片提示词生成器？不止如此。马甲系统 —— 把脑海中的画面，翻译成AI能理解的专业表达。用得越多，它越懂你：首次需要多问几句确认方向，用久了几乎一说就懂。用得越多，它越快：缓存机制让后续对话越来越省。 RAG进化：成功案例持续入库，越跑越聪明。输入「新手指南」查看完整功能介绍

下载

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

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 查找失败。