如何在multiprocessing.Manager对象中创建并更新嵌套字典？

问题重现 #

当使用multiprocessing.Manager.dict()创建共享字典，并在其中嵌套普通字典时，调用嵌套字典的update()、clear()等方法不会同步到共享字典中，示例代码及运行结果如下：

示例代码 #

from multiprocessing import Manager, Process


def worker(shared_object):
    print(f'Before defining the nested dictionary: {shared_object}')
    shared_object['nested'] = {
        'first_key': 'first_value'
    }
    print(f'After defining the nested dictionary: {shared_object}')
    shared_object['nested'].update(
        {
            'second_key': 'second_value'
        }
    )
    print(f'After updating the nested dictionary: {shared_object}')


if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = manager.dict()
        worker_one = Process(target=worker, args=(shared_dict,))
        worker_one.start()
        worker_one.join()

运行结果 #

Before defining the nested dictionary: {}
After defining the nested dictionary: {'nested': {'first_key': 'first_value'}}
After updating the nested dictionary: {'nested': {'first_key': 'first_value'}}

原因分析 #

Manager.dict()返回的是代理对象，它会将对自身的操作同步到进程间的共享内存中。但当你给它赋值一个普通字典（shared_object['nested'] = {...}）时，这个嵌套字典是普通的Python字典，不是代理对象。对这个普通字典调用update()等方法时，操作仅在当前进程的本地副本中生效，不会触发代理对象的同步机制，因此共享字典不会更新。

解决方案 #

方案1：将嵌套字典也创建为Manager代理对象 #

直接用manager.dict()创建嵌套的共享字典，这样嵌套字典本身也是代理对象，内部操作会同步：

from multiprocessing import Manager, Process


def worker(shared_object, manager):
    print(f'Before defining the nested dictionary: {shared_object}')
    # 用manager.dict()创建嵌套的共享字典
    shared_object['nested'] = manager.dict({
        'first_key': 'first_value'
    })
    print(f'After defining the nested dictionary: {shared_object}')
    shared_object['nested'].update(
        {
            'second_key': 'second_value'
        }
    )
    print(f'After updating the nested dictionary: {shared_object}')


if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = manager.dict()
        worker_one = Process(target=worker, args=(shared_dict, manager))
        worker_one.start()
        worker_one.join()

方案2：修改后重新赋值嵌套字典 #

先取出嵌套字典，修改后再重新赋值给共享字典的键，触发代理对象的同步：

from multiprocessing import Manager, Process


def worker(shared_object):
    print(f'Before defining the nested dictionary: {shared_object}')
    shared_object['nested'] = {
        'first_key': 'first_value'
    }
    print(f'After defining the nested dictionary: {shared_object}')
    # 取出嵌套字典，修改后重新赋值
    nested_dict = shared_object['nested']
    nested_dict.update({'second_key': 'second_value'})
    shared_object['nested'] = nested_dict
    print(f'After updating the nested dictionary: {shared_object}')


if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = manager.dict()
        worker_one = Process(target=worker, args=(shared_dict,))
        worker_one.start()
        worker_one.join()

方案3：使用自定义的共享嵌套字典类 #

如果需要频繁操作嵌套字典，可以封装一个类，内部处理代理对象的同步逻辑：

from multiprocessing import Manager, Process


class SharedNestedDict:
    def __init__(self, manager):
        self._dict = manager.dict()
    
    def __getitem__(self, key):
        return self._dict[key]
    
    def __setitem__(self, key, value):
        if isinstance(value, dict):
            value = self._dict._manager.dict(value)
        self._dict[key] = value
    
    def update_nested(self, parent_key, update_dict):
        nested = self._dict[parent_key]
        nested.update(update_dict)
        # 重新赋值触发同步
        self._dict[parent_key] = nested
    
    def __repr__(self):
        return repr(self._dict)


def worker(shared_object):
    print(f'Before defining the nested dictionary: {shared_object}')
    shared_object['nested'] = {'first_key': 'first_value'}
    print(f'After defining the nested dictionary: {shared_object}')
    shared_object.update_nested('nested', {'second_key': 'second_value'})
    print(f'After updating the nested dictionary: {shared_object}')


if __name__ == '__main__':
    with Manager() as manager:
        shared_dict = SharedNestedDict(manager)
        worker_one = Process(target=worker, args=(shared_dict,))
        worker_one.start()
        worker_one.join()

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