如何将multiprocessing队列传递给子进程？附进程容错需求

首先你当前的代码存在核心问题：直接将multiprocessing.Manager().Queue()对象作为命令行参数传给subprocess.Popen是不可行的。命令行参数只能传递字符串，而队列对象无法直接转为可被独立子进程解析的字符串——multiprocessing的队列依赖进程间共享内存或Manager进程的通信机制，subprocess启动的是完全独立的Python进程，和控制器进程没有共享这些底层资源，就算强行序列化队列，子进程也无法恢复出可用的队列实例。

结合你的需求（启动多个子进程、实时检测退出并重启、避免fork复制资源），以下是两种可行方案：

方案一：基于multiprocessing.Manager的跨进程队列通信 #

通过让子进程主动连接到控制器启动的Manager进程，获取共享队列，实现消息传递：

控制器代码 #

import os
import time
import subprocess
from multiprocessing.managers import BaseManager

# 全局队列，供Manager暴露给子进程
exit_queue = None

def get_exit_queue():
    return exit_queue

if __name__ == "__main__":
    # 启动自定义Manager，暴露队列获取方法
    exit_queue = BaseManager().Queue()
    class ExitQueueManager(BaseManager):
        pass
    ExitQueueManager.register('get_exit_queue', callable=get_exit_queue)
    
    # 指定Manager监听地址和认证密钥
    manager = ExitQueueManager(address=('localhost', 50001), authkey=b'child_exit_key')
    manager.start()
    
    # 启动子进程的方法
    def start_child(child_id):
        return subprocess.Popen(
            ['python3', 'multi_child.py', 'localhost', '50001', 'child_exit_key', child_id],
            env=os.environ.copy(),
            stderr=subprocess.PIPE
        )
    
    # 启动两个子进程
    active_processes = {
        'child1': start_child('child1'),
        'child2': start_child('child2')
    }
    
    # 监听队列，处理子进程退出事件
    while True:
        child_name, exit_code = exit_queue.get()
        print(f"子进程 {child_name} 退出，退出码：{exit_code}，正在重启...")
        # 清理旧进程（如果仍存在）
        if active_processes[child_name].poll() is None:
            active_processes[child_name].kill()
        # 启动新进程
        active_processes[child_name] = start_child(child_name)

子进程代码（multi_child.py） #

import sys
import os
import time
from multiprocessing.managers import BaseManager

def main():
    host = sys.argv[1]
    port = int(sys.argv[2])
    authkey = sys.argv[3].encode('utf-8')
    child_name = sys.argv[4]
    
    # 连接到控制器的Manager，获取队列
    class ExitQueueManager(BaseManager):
        pass
    ExitQueueManager.register('get_exit_queue')
    manager = ExitQueueManager(address=(host, port), authkey=authkey)
    manager.connect()
    exit_queue = manager.get_exit_queue()
    
    print(f"子进程 {child_name} ({os.getpid()}) 启动")
    try:
        # 模拟业务逻辑运行
        time.sleep(3)
        # 可主动抛出异常测试容错：raise Exception("模拟崩溃")
        exit_code = 0
    except Exception as e:
        print(f"子进程 {child_name} 异常：{str(e)}")
        exit_code = 1
    finally:
        # 向队列写入退出信息
        exit_queue.put((child_name, exit_code))
        sys.exit(exit_code)

if __name__ == "__main__":
    main()

这种方案利用multiprocessing的成熟通信机制，适合需要传递复杂消息的场景，但需要维护Manager进程，且子进程需知道Manager的地址和密钥。

方案二：基于命名管道的轻量通信 #

如果只需要传递子进程退出信息，命名管道是更简单的选择——无需依赖multiprocessing的Manager，直接通过文件系统实现跨进程通信：

控制器代码 #

import os
import time
import subprocess
import threading

# 命名管道路径（Windows需改为 \\.\pipe\child_exit_pipe）
PIPE_PATH = '/tmp/child_exit_pipe'

# 创建命名管道（仅第一次运行时创建）
if not os.path.exists(PIPE_PATH):
    os.mkfifo(PIPE_PATH)

def listen_pipe():
    """监听命名管道，处理子进程退出信息"""
    with open(PIPE_PATH, 'r') as pipe:
        while True:
            msg = pipe.readline().strip()
            if not msg:
                continue
            child_name, exit_code = msg.split(',')
            exit_code = int(exit_code)
            print(f"子进程 {child_name} 退出，退出码：{exit_code}，正在重启...")
            # 重启对应子进程
            active_processes[child_name] = start_child(child_name)

def start_child(child_name):
    """启动单个子进程"""
    return subprocess.Popen(
        ['python3', 'multi_child.py', child_name, PIPE_PATH],
        env=os.environ.copy(),
        stderr=subprocess.PIPE
    )

if __name__ == "__main__":
    # 启动管道监听线程
    threading.Thread(target=listen_pipe, daemon=True).start()
    
    # 初始化子进程
    active_processes = {
        'child1': start_child('child1'),
        'child2': start_child('child2')
    }
    
    # 控制器保持运行
    while True:
        time.sleep(1)

子进程代码（multi_child.py） #

import sys
import os
import time

def main():
    child_name = sys.argv[1]
    pipe_path = sys.argv[2]
    
    print(f"子进程 {child_name} ({os.getpid()}) 启动")
    try:
        # 模拟业务逻辑运行
        time.sleep(4)
        # 可主动抛出异常测试容错：raise Exception("模拟崩溃")
        exit_code = 0
    except Exception as e:
        print(f"子进程 {child_name} 异常：{str(e)}")
        exit_code = 1
    finally:
        # 向命名管道写入退出信息
        with open(pipe_path, 'w') as pipe:
            pipe.write(f"{child_name},{exit_code}\n")
        sys.exit(exit_code)

if __name__ == "__main__":
    main()

这种方案轻量、易实现，无需额外依赖，适合简单的退出通知场景，跨平台只需调整管道路径格式即可。

内容的提问来源于stack exchange，提问作者Abdullah Chaudhry