Python脚本并发进程管控问题：多进程报错与PyInstaller异常

开发Python脚本实现以下功能：

• 让process1.exe与process2.exe并发运行
• 启动后执行检查及额外代码
• 若process2.exe提前终止，需立即终止process1.exe并停止所有操作
• 检查完成后，等待process2.exe关闭再终止process1.exe

import psutil
import subprocess
import multiprocessing
import os
import time

def findPID():
    process_pid = {}
    process_list = [p.info for p in psutil.process_iter(['name', 'pid'])]
    
    for proc in process_list:
        if proc['name'] == "process2.exe" or proc['name'] == "process1.exe":
            process_name = proc['name']
            process_pid[f"{process_name}"] = proc["pid"]

    return process_pid


def killSimunteneously(event, queue, running_process, process_pid):
    while True:
        if not psutil.pid_exists(process_pid['process2.exe']):
            try:
                process_to_kill = psutil.Process(process_pid['process1.exe'])
                process_to_kill.terminate()
                break
            except:
                break
            
        else:
            time.sleep(2)
    
    queue.put("killSimunteneously Completed!")
    event.set()

    if running_process.is_alive():
        print("Terminating checkStatus")
        running_process.terminate()

def checkStatus(event, queue):
    # Some Code

    queue.put("checkStatus Completed!")

    if not event.is_set():
        print("Waiting for killSimunteneously")
        event.wait()
    
def main():
    os.startfile(".\process2.exe")
    
    subprocess.Popen([".\process1.exe"], creationflags=subprocess.CREATE_NEW_CONSOLE)
    
    bg_process_PID = findPID()
    
    queue = multiprocessing.Queue()
    event = multiprocessing.Event()

    p1 = multiprocessing.Process(target=checkStatus, args=(event, queue))
    p1.start()

    p2 = multiprocessing.Process(target=killSimunteneously, args=(event, queue, p1, bg_process_PID,))
    p2.start()

    p1.join()
    p2.join()

    while not queue.empty():
        print(queue.get())

if __name__ == "__main__":
    main()

Traceback (most recent call last):
  File "D:\codes\script.py", line 71, in <module>
    main()
  File "D:\codes\script.py", line 62, in main
    p2.start()
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\process.py", line 121, in start
    self._popen = self._Popen(self)
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\context.py", line 224, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\context.py", line 336, in _Popen
    return Popen(process_obj)
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\popen_spawn_win32.py", line 93, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle 'weakref.ReferenceType' object
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "C:\Users\ADMIN\AppData\Local\Programs\Python\Python310\lib\multiprocessing\spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
EOFError: Ran out of input

• 不使用multiprocessing.Process()，顺序执行checkStatus()和killSimultaneously()时功能正常，但process2.exe提前终止时无法及时终止process1.exe
• 最初未使用队列和事件时，PyInstaller生成的exe会重复启动process2.exe三次，而.py文件运行正常

错误原因 #

报错核心是无法序列化multiprocessing.Process对象：你把p1（一个Process实例）作为参数传给了p2进程，而Process对象内部包含弱引用（weakref），这类对象无法被pickle序列化，跨进程传递时就会抛出TypeError。

修复方案 #

1. 移除跨进程传递Process对象的逻辑 #

让主进程负责管理子进程的生命周期，不需要在killSimunteneously里直接操作p1。可以通过Event来通知checkStatus进程自行退出，或者由主进程在合适时机终止子进程。

2. 优化PID获取逻辑 #

启动process1和process2后立即调用findPID可能因为进程尚未完全启动导致找不到PID，需要添加短暂延迟和重试机制。

3. 修复PyInstaller重复启动问题 #

在主入口处添加multiprocessing.freeze_support()，避免spawn模式下的重复执行。

修复后的完整代码 #

import psutil
import subprocess
import multiprocessing
import os
import time

def findPID():
    process_pid = {}
    # 重试3次，避免进程未启动完成
    for _ in range(3):
        process_list = [p.info for p in psutil.process_iter(['name', 'pid'])]
        for proc in process_list:
            if proc['name'] == "process2.exe" or proc['name'] == "process1.exe":
                process_name = proc['name']
                process_pid[f"{process_name}"] = proc["pid"]
        if len(process_pid) == 2:
            break
        time.sleep(1)
    return process_pid


def monitor_process2(event, queue, process_pid):
    while True:
        if not psutil.pid_exists(process_pid.get('process2.exe', 0)):
            try:
                # 终止process1
                if 'process1.exe' in process_pid and psutil.pid_exists(process_pid['process1.exe']):
                    process_to_kill = psutil.Process(process_pid['process1.exe'])
                    process_to_kill.terminate()
                    print("process1.exe已终止")
                break
            except Exception as e:
                print(f"终止process1失败: {e}")
                break
        time.sleep(2)
    
    queue.put("process2监控完成!")
    event.set()

def checkStatus(event, queue):
    # 这里放你的检查代码
    print("执行检查逻辑...")
    time.sleep(5)  # 模拟检查耗时
    
    queue.put("检查完成!")

    if not event.is_set():
        print("等待process2关闭...")
        event.wait()
    
def main():
    # 启动process2
    os.startfile(".\\process2.exe")
    # 启动process1
    subprocess.Popen([".\\process1.exe"], creationflags=subprocess.CREATE_NEW_CONSOLE)
    
    # 等待进程启动后获取PID
    time.sleep(1)
    bg_process_PID = findPID()
    if len(bg_process_PID) != 2:
        print("无法获取process1或process2的PID")
        return
    
    queue = multiprocessing.Queue()
    event = multiprocessing.Event()

    p1 = multiprocessing.Process(target=checkStatus, args=(event, queue))
    p1.start()

    # 不再传递p1进程对象，改用event控制
    p2 = multiprocessing.Process(target=monitor_process2, args=(event, queue, bg_process_PID,))
    p2.start()

    # 主进程等待子进程结束
    p1.join()
    p2.join()

    # 输出队列信息
    while not queue.empty():
        print(queue.get())

if __name__ == "__main__":
    multiprocessing.freeze_support()
    main()

修复说明 #

• 移除了跨进程传递Process对象的操作，改用Event同步两个子进程的状态
• 优化findPID函数，增加重试机制确保能获取到进程PID
• 添加multiprocessing.freeze_support()解决PyInstaller打包后的重复启动问题
• 重命名killSimunteneously为monitor_process2，明确函数职责

内容的提问来源于stack exchange，提问作者Siddharth Mandaliya