Python多进程场景下子进程创建的Queue如何共享给主进程

你遇到的报错本质是：原生multiprocessing.Queue仅支持通过进程创建时的参数继承传递，无法被multiprocessing.Manager序列化后存入托管容器，只有Manager实例自身创建的托管Queue对象支持跨进程存入共享列表。

你的约束是Queue必须在子进程内部生成，只要将主进程创建的Manager实例传递给子进程，子进程内部调用Manager的Queue()方法生成队列，再存入共享列表即可满足要求。
修改后的可运行代码如下：

import multiprocessing as mp
from time import sleep

class Test:
    def __init__(self, manager, qlist):
        self.manager = manager
        self.qlist = qlist
        # 子进程内部生成托管Queue，可存入共享列表
        self.q = manager.Queue()
        # 把当前子进程的Queue追加到共享列表
        qlist.append(self.q)
        self.run()

    def run(self):
        while True:
            val = self.q.get()
            if val == 1:
                # 新建子进程，传递manager和共享qlist
                p = mp.Process(target=Test, args=(self.manager, self.qlist))
                p.start()
            else:
                print(f"子进程收到消息：{val}")

if __name__ == '__main__':
    manager = mp.Manager()
    qlist = manager.list()

    # 启动第一个子进程
    p = mp.Process(target=Test, args=(manager, qlist))
    p.start()

    sleep(0.5)
    # 测试：给第一个子进程发1，触发新建子进程
    qlist[0].put(1)
    sleep(0.5)
    # 现在qlist里有2个Queue，给第二个发消息
    qlist[1].put("Hello")
    sleep(0.5)
    # 打印当前共享列表长度，验证新建成功
    print(f"当前共享列表内Queue数量：{len(qlist)}")

如果你需要直接共享包含self.q属性的Test对象，可以通过注册自定义类到Manager实现：

• 调用mp.managers.BaseManager.register('Test', Test)把Test类注册到管理器
• 子进程内部通过manager.Test()创建实例，再存入共享列表即可
  主进程可直接调用共享列表中Test实例的q.put()方法发送消息，完全符合你的需求。

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