本文介绍一种简洁可靠的方案,使用 subprocess 与 multiprocessing 协同管理两个外部可执行文件(如 process1.exe 和 process2.exe),确保 process1 在 process2 异常退出或正常结束时被及时终止,并避免进程泄漏、序列化错误及重复启动问题。
在实际的自动化脚本开发中,我们常常需要并行启动多个外部程序,比如仿真器、服务进程,并且要求它们之间存在严格的依赖关系——process1 的生命周期必须完全受控于 process2。换句话说,process2 一旦退出,无论成功还是崩溃,process1 必须立即终止,主脚本也要同步停止后续检查逻辑。这个需求听起来简单,但实现起来坑不少。很多开发者最初会尝试用 psutil 轮询 PID 配合多进程监听,结果在 Windows 下碰上 TypeError: cannot pickle 'weakref.ReferenceType' object——这是 multiprocessing 使用 spawn 启动方式时的典型限制,因为线程池不能序列化那些包含弱引用或文件句柄的对象。
那么,问题根源在哪?核心原则其实就一条:不要将 subprocess.Popen 实例、psutil.Process 对象或任何含弱引用/文件句柄的资源跨进程传递。正确的做法是:由主进程直接持有并管理子进程句柄,利用阻塞式等待(.wait())和信号机制(Event)实现轻量级协同,而非依赖多进程轮询 PID。下面直接给出优化后的完整实现,代码里每一处都有注释,方便对照理解。
import subprocess
import multiprocessing
import sys
def checkStatus(event, queue):
"""
执行关键检查逻辑(如日志扫描、端口探测、状态码验证等)
注意:该函数运行在独立进程中,不可直接访问主进程的 subprocess.Popen 对象
"""
try:
# 示例检查逻辑(请按需替换)
print("→ 开始执行前置检查...")
for i in range(3):
if event.is_set(): # 检测 process2 是否已退出
print("⚠️ process2 已退出,中止检查")
break
print(f" 检查步骤 {i+1}/3...")
# time.sleep(1) # 实际场景中可加入真实校验逻辑
else:
print("✅ 检查完成,等待 process2 结束...")
queue.put("checkStatus Completed!")
except Exception as e:
queue.put(f"checkStatus Error: {e}")
finally:
# 确保事件被响应(即使异常也通知主进程)
if not event.is_set():
event.set()
def main():
# ✅ 正确启动方式:主进程直接持有 Popen 实例
try:
process1 = subprocess.Popen(
["./process1.exe"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
creationflags=subprocess.CREATE_NO_WINDOW # 隐藏控制台(可选)
)
process2 = subprocess.Popen(
["./process2.exe"],
creationflags=subprocess.CREATE_NEW_CONSOLE # 显示独立窗口
)
except FileNotFoundError as e:
print(f"❌ 启动失败:{e}")
return
except Exception as e:
print(f"❌ 进程启动异常:{e}")
return
# ? 多进程仅用于异步检查,不参与进程控制
queue = multiprocessing.Queue()
event = multiprocessing.Event()
p_check = multiprocessing.Process(target=checkStatus, args=(event, queue))
p_check.start()
try:
# ⏳ 主进程阻塞等待 process2 结束(天然支持异常退出检测)
print("⏳ 主进程等待 process2.exe 完成...")
process2.wait() # 若 process2 崩溃,此调用立即返回
print("✅ process2.exe 已退出")
# ? 立即终止 process1(无论其当前状态)
if process1.poll() is None: # 仍运行中
process1.terminate()
try:
process1.wait(timeout=5) # 等待优雅退出
except subprocess.TimeoutExpired:
process1.kill() # 强制终止
print("❗ process1.exe 被强制杀死")
else:
print("ℹ️ process1.exe 已自行退出")
# ? 通知检查进程终止
event.set()
# ? 清理检查子进程
p_check.join(timeout=2)
if p_check.is_alive():
p_check.terminate()
p_check.join()
# ? 输出检查结果
while not queue.empty():
print(queue.get())
except KeyboardInterrupt:
print("n? 用户中断,正在清理...")
process1.terminate()
process2.terminate()
p_check.terminate()
p_check.join()
finally:
# ? 关键:确保资源释放(尤其 Windows 下句柄泄漏风险)
if 'process1' in locals():
process1.stdout.close() if process1.stdout else None
if __name__ == "__main__":
main()
✅ 关键改进说明
- 消除序列化风险:不再向子进程传递 psutil.Process 或 Popen 实例,所有进程控制均由主进程统一调度。
- 精准生命周期绑定:
process2.wait()是最可靠的方式监听其退出(包括崩溃),无需轮询 PID。 - 健壮的异常处理:涵盖 FileNotFoundError(路径错误)、TimeoutExpired(强制杀进程)、KeyboardInterrupt(用户中止)。
- 资源安全释放:显式关闭 stdout,避免 Windows 下句柄泄漏;使用 terminate() + wait(timeout) + kill() 三级终止策略。
- PyInstaller 兼容性:避免
os.startfile()(易触发多实例),改用 subprocess.Popen,确保打包后行为一致。
⚠️ 注意事项
- 若 process2.exe 是 GUI 程序且无控制台,
CREATE_NEW_CONSOLE可省略;若需隐藏窗口,改用CREATE_NO_WINDOW(Windows)或start_new_session=True(Linux/macOS)。 - checkStatus 中的业务逻辑应避免长时间阻塞,建议添加超时或定期检查
event.is_set()。 - 生产环境建议增加日志记录(如 logging 模块)替代 print,便于追踪进程状态。
这个方案以最小复杂度达成高可靠性,兼顾可读性与工程实践要求。既不会因为序列化错误而翻车,也不会因为轮询 PID 导致 CPU 空转。如果拖到生产环境,记得把 print 换成 logging,再补上超时保护,基本就稳了。