无法将对象存入Python multiprocessing共享字典的问题解决

错误原因 #

1. 自定义类序列化失败：子进程启动时会重新导入主模块，若主程序逻辑未包裹在if __name__ == '__main__':块中，子进程无法正确加载MyObj类，导致反序列化对象时出现属性找不到的错误。
2. 共享对象访问异常：直接依赖全局的dictionary在子进程中无法正确关联Manager创建的共享字典，需通过进程池初始化函数将共享对象明确传入子进程。
3. 冗余代码干扰：原代码中的output_queue和task_init未实际发挥作用，属于无效代码。

修正后的实现代码 #

import multiprocessing as mp
from functools import partial

class MyObj:
    def __init__(self, x):
        self.x = x  # 修正原代码中参数未赋值的问题

def task(lock, shared_dict, obj):
    with lock:  # 用with语句自动管理锁，避免手动操作出错
        shared_dict[f"obj_{obj.x}"] = obj  # 用对象属性做键，避免值被覆盖

def init_worker(shared_dict, lock):
    # 将共享对象绑定到子进程全局变量
    global dictionary, task_lock
    dictionary = shared_dict
    task_lock = lock

if __name__ == '__main__':
    manager = mp.Manager()
    shared_dict = manager.dict()
    lock = manager.Lock()

    # 生成测试对象列表
    my_list = [MyObj(i) for i in range(3)]

    # 初始化进程池并传入共享资源
    with mp.Pool(mp.cpu_count(), init_worker, [shared_dict, lock]) as p:
        func = partial(task, task_lock, shared_dict)
        p.map(func, my_list)

    # 验证共享字典结果
    print("共享字典内容：")
    for key, obj in shared_dict.items():
        print(f"{key}: x = {obj.x}")
    print("finished")

关键修改说明 #

• 主逻辑包裹进if __name__ == '__main__':：避免子进程重复执行主代码，确保MyObj类能被正确加载。
• 进程池初始化传递共享对象：通过init_worker把Manager创建的字典和锁传入每个子进程，保证所有进程访问的是同一共享资源。
• 自动管理锁：用with语句替代手动acquire/release，防止异常导致锁无法释放。
• 修正对象初始化逻辑：让MyObj的参数x正确赋值给实例属性，使对象具备实际可识别的特征。
• 自动管理进程池：用with语句自动完成进程池的关闭和回收，简化代码流程。

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