使用multiprocessing.Manager跨进程同步列表字典的规则解析

问题背景 #

使用multiprocessing模块进行多核CPU密集型任务处理时，进程间传递的列表、字典经常出现修改不同步的情况：比如调用append/extend修改共享列表时同步生效，但直接将共享列表赋值为新列表时同步失效；嵌套结构的深度修改也无法同步到其他进程。

以下是最小可复现示例：

from time import sleep
from multiprocessing import Manager, Process

def parent_func(shared_list):
        sleep(1)
        shared_list.extend(['a','b','c'])  # 修改会同步到其他进程
        # shared_list = ['a','b','c']  # 直接赋值，同步失效

def child_func(shared_list):
    k = 0
    while k < 8:
        k += 1
        print(f'{k}: {shared_list}')
        sleep(0.2)

def main_func():
    with Manager() as manager:
        shared_list = manager.list()

        process = Process(target=parent_func, args=(shared_list,))
        processes = [process]
        process.start()
        process = Process(target=child_func, args=(shared_list,))
        processes.append(process)
        process.start()

        for process in processes:
            process.join()

        print('---')
        print(list(shared_list))

if __name__ == '__main__':
    main_func()

同步核心规则 #

multiprocessing Manager创建的manager.list()/manager.dict()本质是代理对象，通过远程过程调用（RPC）实现跨进程同步，核心规则如下：

• 仅当调用代理对象的内置方法（如列表的append/extend/pop，字典的update/popitem）或直接修改元素（如shared_list[0] = 1、shared_dict['key'] = 'val'）时，才会触发同步逻辑，修改会传递到其他进程。
• 若直接给代理变量重新赋值（如shared_list = ['a','b','c']），本质是将当前进程的局部变量指向了一个新的原生容器，完全切断了与原代理对象的关联，修改自然无法同步。

嵌套容器的深度同步实现 #

默认的代理对象仅对顶层容器的操作同步，嵌套的原生容器（如代理列表里的普通字典）修改不会触发同步。要实现深度同步，需保证每层嵌套容器都是Manager创建的代理对象：

示例代码 #

from time import sleep
from multiprocessing import Manager, Process

def update_nested(shared_nested):
    sleep(1)
    # 嵌套字典是manager创建的代理，修改会同步
    shared_nested[0]['name'] = 'test'
    # 新增嵌套元素时，也要用manager创建代理
    shared_nested.append(manager.dict({'age': 20}))

def read_nested(shared_nested):
    k = 0
    while k < 8:
        k += 1
        print(f'{k}: {shared_nested}')
        sleep(0.2)

def main():
    with Manager() as manager:
        # 顶层列表和内部字典都是代理对象
        shared_nested = manager.list([manager.dict()])
        p1 = Process(target=update_nested, args=(shared_nested,))
        p2 = Process(target=read_nested, args=(shared_nested,))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        print('---')
        print(list(shared_nested))

if __name__ == '__main__':
    main()

关键注意点 #

• 所有需要跨进程同步修改的嵌套容器，必须通过manager.list()/manager.dict()创建，不能使用原生的[]/{}。
• 若在代理容器中新增元素，新增的元素也必须是Manager代理对象，否则后续对该元素的修改无法同步。

总结 #

1. 避免直接给Manager代理变量重新赋值，仅通过其内置方法或元素赋值操作修改内容。
2. 嵌套容器需逐层使用Manager创建代理，确保每一层的修改都能触发同步逻辑。
3. 原生容器无法跨进程同步，所有共享容器必须是Manager生成的代理对象。

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