本文介绍如何在 python 中可靠地并发启动两个外部程序（如 process1.exe 和 process2.exe），实现进程间状态联动——当 process2 异常退出时立即终止 process1，或在其正常结束后优雅关闭 process1，并确保主流程同步协调不出现 pickle 错误或资源竞争。
本文介绍如何在 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 提供的可靠进程控制能力，逻辑清晰、调试友好、部署稳定，是生产环境中管理外部进程协作的推荐范式。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
