多进程Pool映射对象后无法感知状态修改，如何实现预期效果？

我创建了一个对象列表，通过multiprocessing.Pool并行对每个对象执行修改其状态的任务。进程池任务执行完成后，遍历对象并打印修改结果，却仍显示初始状态。请问如何实现预期的状态修改效果？

import multiprocessing as mp

class MyObj:
    def __init__(self, x):
        self.x = 0

def task_init(output_queue):
    task.output_queue = output_queue

def task(obj):
    obj.x = 5       # 这里尝试修改值

my_list = []

obj = MyObj("")
my_list.append(obj)

output_queue = mp.Queue()

p = mp.Pool(mp.cpu_count(), task_init, [output_queue])

p.map(task, my_list)

for obj in my_list:
    print(str(obj.x)) # 问题：显示初始值0，而非预期的5

原因分析 #

Python的multiprocessing模块创建的子进程拥有独立的内存空间，当你把对象传递给子进程的任务时，实际上传递的是原对象的副本（通过pickle序列化/反序列化实现）。子进程修改的只是这个副本的状态，完全不会影响主进程中原始对象的值。

解决方案 #

方案1：让任务返回修改后的对象，替换原列表元素 #

直接让task函数返回修改后的对象，然后用Pool.map()的返回结果覆盖原列表。这样主进程就能拿到子进程修改后的对象。

修改后的代码：

import multiprocessing as mp

class MyObj:
    def __init__(self, x):
        self.x = 0

def task(obj):
    obj.x = 5
    return obj  # 返回修改后的对象

my_list = [MyObj("")]

with mp.Pool(mp.cpu_count()) as p:
    # 用map的返回结果替换原列表
    my_list = p.map(task, my_list)

for obj in my_list:
    print(obj.x)  # 输出5，符合预期

方案2：使用共享内存对象 #

如果需要在多个进程间共享对象状态，可以使用multiprocessing.Manager创建可共享的对象，或者用Value/Array来管理单个共享属性。

示例（用Manager创建可共享的自定义对象）：

import multiprocessing as mp

class MyObj:
    def __init__(self):
        self.x = mp.Value('i', 0)  # 用共享整数类型存储x

def task(obj):
    obj.x.value = 5  # 修改共享变量的值

if __name__ == '__main__':
    with mp.Manager() as manager:
        # 创建可被多进程共享的MyObj实例
        shared_obj = manager.Namespace()
        shared_obj.obj = MyObj()
        
        my_list = [shared_obj.obj]
        
        with mp.Pool(mp.cpu_count()) as p:
            p.map(task, my_list)
        
        print(shared_obj.obj.x.value)  # 输出5，符合预期

方案3：使用队列传递修改结果 #

利用multiprocessing.Queue，让子进程把修改后的对象或属性值发送回主进程，主进程再更新原列表。

示例：

import multiprocessing as mp

class MyObj:
    def __init__(self, idx):
        self.idx = idx
        self.x = 0

def task(args):
    obj, queue = args
    obj.x = 5
    queue.put((obj.idx, obj))  # 把索引和修改后的对象放入队列

if __name__ == '__main__':
    my_list = [MyObj(0)]
    output_queue = mp.Queue()
    
    with mp.Pool(mp.cpu_count()) as p:
        # 传递对象和队列作为任务参数
        p.map(task, [(obj, output_queue) for obj in my_list])
    
    # 从队列取出结果，更新原列表
    while not output_queue.empty():
        idx, updated_obj = output_queue.get()
        my_list[idx] = updated_obj
    
    for obj in my_list:
        print(obj.x)  # 输出5，符合预期

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