Python多进程内存共享实现及大字典预加载操作方法咨询

你当前写法性能差的核心原因是多进程默认的任务传参机制：multiprocessing会对每个任务的所有参数做pickle序列化，再拷贝到子进程。大字典的每个value都要重复序列化、传输，当拷贝开销超过多进程计算收益时，速度就会比单进程更慢。

如果你的大字典是只读不需要跨进程修改的场景，优先用这个方案，性能远高于共享内存方案：

• 利用Pool的initializer参数，仅在每个子进程创建时执行一次字典加载，后续所有任务都可以复用预加载的字典，不需要每次传参
• Unix/Linux/macOS下利用fork的Copy-On-Write机制，子进程只要不修改字典，就不会发生内存拷贝，完全共享父进程的内存空间，没有额外内存占用
• 符合你多次复用子进程的需求，同一个进程池的多次map/apply调用都可以直接用预加载的字典

实现代码 #

import multiprocessing as mp

# 子进程全局变量，用于存储预加载的共享数据
shared_big_dict: dict = None
shared_common_params: any = None

def worker_init(big_dict: dict, common_params: any):
    """子进程初始化函数，每个子进程仅在创建时执行1次"""
    global shared_big_dict, shared_common_params
    shared_big_dict = big_dict
    shared_common_params = common_params

def worker_task(k: any):
    """工作函数，直接读取预加载的全局字典，无需额外传参"""
    v = shared_big_dict[k]
    # 这里替换为你原来的f函数处理逻辑
    processed_result = v # 示例，替换为实际处理结果
    return k, processed_result

if __name__ == "__main__":
    # 你的大嵌套字典
    x = {"key1": {"subkey1": 1}, "key2": {"subkey2": 2}}
    # 其他固定的公共参数
    something = "你的其他公共参数"
    
    # 创建进程池时指定初始化函数和参数
    with mp.Pool(initializer=worker_init, initargs=(x, something)) as p:
        # 仅传递key即可，序列化开销极低
        result_list = p.map(worker_task, x.keys())
    # 转换为结果字典
    y = dict(result_list)

如果需要多进程修改字典内容，再用Manager实现跨进程共享，注意嵌套字典需要递归转换为Manager的DictProxy对象才能实现全层级共享。

实现代码 #

import multiprocessing as mp
from multiprocessing.managers import DictProxy

def convert_to_manager_dict(raw_dict: dict, manager: mp.Manager) -> DictProxy:
    """递归将普通嵌套字典转为可跨进程共享的Manager字典"""
    m_dict = manager.dict()
    for k, v in raw_dict.items():
        if isinstance(v, dict):
            m_dict[k] = convert_to_manager_dict(v, manager)
        else:
            m_dict[k] = v
    return m_dict

def worker_task(args: tuple):
    k, shared_dict, common_params = args
    v = shared_dict[k]
    # 这里可以读写shared_dict，修改会同步到所有进程
    # 示例：shared_dict[k]["new_subkey"] = "test"
    processed_result = v # 替换为实际处理结果
    return k, processed_result

if __name__ == "__main__":
    x = {"key1": {"subkey1": 1}, "key2": {"subkey2": 2}}
    something = "你的其他公共参数"
    
    with mp.Manager() as manager:
        # 转换为可共享的嵌套字典
        shared_x = convert_to_manager_dict(x, manager)
        with mp.Pool() as p:
            result_list = p.map(worker_task, ((k, shared_x, something) for k in x.keys()))
        y = dict(result_list)
        # 所有子进程对shared_x的修改都会保留在这里

• 只读场景100%优先用预加载方案，性能比Manager方案高1~2个数量级，没有额外内存开销
• Manager方案仅适用于需要跨进程修改共享字典的场景，每次读写都有跨进程通信开销，高频读写场景不建议使用

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