使用multiprocessing时无法保存实例变量的原因及解决方法

先看你的代码：

from multiprocessing import Process


class B:
    def __init__(self, a, b):
        self.a = a
        self.b = b


class A:
    def __init__(self, a):
        self.k = None
        self.k2 = None
        p1 = Process(target=self.a1, args=(a, 1,), daemon=True)
        p2 = Process(target=self.a2, args=(a, 2,), daemon=True)

        p1.start()
        p2.start()

        p1.join()
        p2.join()

    def a1(self, a, b):
        self.k = B(a, b)

    def a2(self, a, b):
        self.k2 = B(a, b)


if __name__ == '__main__':
    t = A(0)
    print(t.k)

问题原因 #

Python多进程的核心机制是内存空间完全独立：

• 启动子进程时，操作系统会复制主进程的所有内存数据（包括A类的实例t），生成一个完全独立的副本。
• 子进程里调用self.a1时，修改的是子进程内存中的t副本，和主进程里的原t实例没有任何关联。
• 子进程结束后，副本内存被销毁，主进程的t.k自然还是初始的None。

修复方案 #

针对你需要并行创建大体积B类实例的需求，推荐以下几种实用方法：

方案1：用队列传递结果（最常用） #

multiprocessing.Queue是多进程间安全传递数据的标准工具，适合返回创建好的B实例：

from multiprocessing import Process, Queue


class B:
    def __init__(self, a, b):
        self.a = a
        self.b = b


class A:
    def __init__(self, a):
        self.k = None
        self.k2 = None
        # 创建两个队列分别接收子进程的结果
        q1 = Queue()
        q2 = Queue()

        p1 = Process(target=self.a1, args=(a, 1, q1), daemon=True)
        p2 = Process(target=self.a2, args=(a, 2, q2), daemon=True)

        p1.start()
        p2.start()

        # 从队列取出结果，赋值给主进程的实例属性
        self.k = q1.get()
        self.k2 = q2.get()

        p1.join()
        p2.join()

    def a1(self, a, b, queue):
        # 将创建好的B实例放入队列
        queue.put(B(a, b))

    def a2(self, a, b, queue):
        queue.put(B(a, b))


if __name__ == '__main__':
    t = A(0)
    print(t.k.a, t.k.b)  # 输出 0 1

方案2：用Manager创建可共享对象 #

如果需要让子进程直接修改主进程的属性，可以用Manager创建可代理的共享字典：

from multiprocessing import Process, Manager


class B:
    def __init__(self, a, b):
        self.a = a
        self.b = b


class A:
    def __init__(self, a):
        # 使用Manager创建可跨进程共享的字典
        with Manager() as manager:
            self.shared_attrs = manager.dict({'k': None, 'k2': None})
            
            p1 = Process(target=self.a1, args=(a, 1, self.shared_attrs), daemon=True)
            p2 = Process(target=self.a2, args=(a, 2, self.shared_attrs), daemon=True)

            p1.start()
            p2.start()
            p1.join()
            p2.join()
            
            # 同步共享字典的值到实例属性
            self.k = self.shared_attrs['k']
            self.k2 = self.shared_attrs['k2']

    def a1(self, a, b, shared_attrs):
        shared_attrs['k'] = B(a, b)

    def a2(self, a, b, shared_attrs):
        shared_attrs['k2'] = B(a, b)


if __name__ == '__main__':
    t = A(0)
    print(t.k.a, t.k.b)  # 输出 0 1

方案3：用ProcessPoolExecutor简化代码 #

如果只是并行执行任务并获取结果，concurrent.futures.ProcessPoolExecutor的接口更简洁：

from concurrent.futures import ProcessPoolExecutor


class B:
    def __init__(self, a, b):
        self.a = a
        self.b = b


class A:
    def __init__(self, a):
        self.k = None
        self.k2 = None
        
        with ProcessPoolExecutor(max_workers=2) as executor:
            # 提交两个创建B的任务
            future1 = executor.submit(self.create_b, a, 1)
            future2 = executor.submit(self.create_b, a, 2)
            
            # 获取执行结果
            self.k = future1.result()
            self.k2 = future2.result()

    @staticmethod
    def create_b(a, b):
        # 抽成静态方法，避免传递整个实例
        return B(a, b)


if __name__ == '__main__':
    t = A(0)
    print(t.k.a, t.k.b)  # 输出 0 1

注意事项 #

• B类需要是可序列化的（默认只要没有文件句柄、线程等不可序列化属性就没问题），因为多进程间传递数据会用pickle序列化。
• 如果B的计算过程涉及共享数据，记得用multiprocessing.Lock等同步工具避免竞争问题。

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