Python多进程中run函数设置的对象属性值不生效的原因及解决方法

问题复现代码 #

from multiprocessing import Process
from multiprocessing import Value
class TestMultiprocess(Process):
  def __init__(self):
    super().__init__()
    self.variable={"value":"initial"}
  def run(self):
    print("executing the run function")
    self.variable={"value":"new"}
    

t=TestMultiprocess()
t.start()
t.join()
print(t.variable) #prints {"value":"initial"}, (I expected {"value":"new"})

问题描述 #

继承自multiprocessing.Process的TestMultiprocess类，在run函数中修改的self.variable属性值，主进程打印时仍为构造函数初始化的{"value":"initial"}，而非预期的{"value":"new"}。需要明确该现象的原因，以及如何在这类子类的run函数中修改字典类型的属性，让主进程能获取到修改后的值。

现象原因 #

多环境下每个进程拥有独立的内存空间。当调用start()启动子进程时，主进程的对象属性会被完整复制一份到子进程的内存中。run方法里修改的self.variable是子进程内存中的副本，和主进程里的原属性是完全独立的两个对象，因此主进程打印的始终是初始值。

字典属性的同步解决方法 #

由于multiprocessing.Value不支持字典类型，可通过以下几种方式实现跨进程的字典属性同步：

方法1：使用multiprocessing.Manager创建共享字典 #

multiprocessing.Manager会启动一个后台管理进程，协调不同进程对共享对象的访问，其中的dict()方法可直接创建跨进程共享的字典。

修改后的代码示例：

from multiprocessing import Process, Manager

class TestMultiprocess(Process):
    def __init__(self):
        super().__init__()
        # 创建共享字典
        self.variable = Manager().dict({"value":"initial"})
    
    def run(self):
        print("executing the run function")
        # 直接修改共享字典的值
        self.variable["value"] = "new"

t = TestMultiprocess()
t.start()
t.join()
# 转换为普通字典后打印
print(dict(t.variable))  # 输出 {"value":"new"}

方法2：使用队列传递修改后的字典 #

利用multiprocessing.Queue作为进程间的数据通道，子进程将修改后的字典放入队列，主进程从队列中取出数据。

代码示例：

from multiprocessing import Process, Queue

class TestMultiprocess(Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue
        self.variable = {"value":"initial"}
    
    def run(self):
        print("executing the run function")
        self.variable["value"] = "new"
        # 将修改后的字典放入队列
        self.queue.put(self.variable)

# 创建队列实例
q = Queue()
t = TestMultiprocess(q)
t.start()
t.join()
# 主进程从队列获取结果
modified_var = q.get()
print(modified_var)  # 输出 {"value":"new"}

方法3：序列化字典到共享内存 #

通过pickle或json将字典序列化为字节串，存储到multiprocessing.Value创建的共享内存中，读取时再反序列化还原字典。

代码示例：

from multiprocessing import Process, Value
import pickle

class TestMultiprocess(Process):
    def __init__(self, shared_data):
        super().__init__()
        self.shared_data = shared_data
    
    def run(self):
        print("executing the run function")
        # 反序列化获取原始字典
        var = pickle.loads(self.shared_data.value)
        var["value"] = "new"
        # 序列化后存回共享内存
        self.shared_data.value = pickle.dumps(var)

# 初始化共享内存，存储序列化后的初始字典
initial_dict = {"value":"initial"}
shared_data = Value('b', pickle.dumps(initial_dict))

t = TestMultiprocess(shared_data)
t.start()
t.join()

# 主进程反序列化获取修改后的值
result = pickle.loads(shared_data.value)
print(result)  # 输出 {"value":"new"}

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