Python中使用multiprocessing模块共享复杂对象的方案

先跟你说透核心问题：你遇到的慢100倍的坑，完全是因为默认的spawn启动方式（Windows强制用，macOS现在默认也是）会把整个复杂对象用pickle序列化后，完整复制给每个子进程——嵌套的列表、元组、字典序列化/反序列化+内存复制的开销，直接把并行的收益啃得一干二净，甚至比单进程还慢。

下面分场景给你最实用的解决方案，都是我踩过坑后验证有效的：

1. Unix系统（Linux/macOS）：直接用fork的写时复制（COW） #

如果你的代码跑在Linux或者老版本macOS上，这是零成本的最优解。

当用multiprocessing.Process时，把启动方式设为fork（Linux默认就是这个，新版macOS需要显式指定），子进程会直接共享父进程的内存空间。因为你的对象是只读的，完全不会触发内存复制——只有当某个进程尝试修改对象时，系统才会复制对应的内存页，而你这里只读取，所以全程零额外开销。

给你写个极简示例：

import multiprocessing as mp

# 定义你的复杂只读对象
class ComplexObj:
    def __init__(self):
        self.nested_list = [{"id": i, "values": tuple(range(100))} for i in range(1000)]
        self.large_dict = {f"item_{i}": {"data": i*2, "tags": ("read-only", "cpu-heavy")} for i in range(5000)}

def cpu_task(obj, task_num):
    # 只做读取操作，绝不修改obj
    total = 0
    for item in obj.nested_list:
        total += sum(item["values"])
    for val in obj.large_dict.values():
        total += val["data"]
    print(f"Task {task_num} finished, total: {total}")

if __name__ == "__main__":
    # 主进程提前创建好复杂对象
    shared_obj = ComplexObj()

    # 显式设置启动方式为fork（macOS需要，Linux可省略）
    mp.set_start_method("fork")

    # 启动4个进程并行处理
    processes = [mp.Process(target=cpu_task, args=(shared_obj, i)) for i in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

这个方案跑起来速度和单进程比，绝对是倍数级提升，完全不用折腾序列化的事。

2. 跨平台方案（含Windows）：用shared_memory提取核心数据 #

如果必须兼容Windows，fork用不了，那得换个思路：把复杂对象里的核心可计算数据提取出来，放到共享内存里，让子进程直接从共享内存加载，而不是复制整个对象。

核心思路是：把嵌套结构里的数值型、可序列化的核心数据，转换成numpy数组这类连续内存结构，然后用multiprocessing.shared_memory创建共享内存块，子进程通过内存名直接访问。

举个可运行的例子：

import multiprocessing as mp
from multiprocessing import shared_memory
import numpy as np

class ComplexObj:
    def __init__(self, shm_name=None):
        if shm_name is None:
            # 主进程：创建数据并写入共享内存
            # 提取嵌套结构里的核心数值
            list_values = np.array([sum(item["values"]) for item in [{"id": i, "values": tuple(range(100))} for i in range(1000)]], dtype=np.int64)
            dict_values = np.array([val["data"] for val in {f"item_{i}": {"data": i*2, "tags": ("read-only", "cpu-heavy")} for i in range(5000)}.values()], dtype=np.int64)
            
            # 拼接成一个大数组，存入共享内存
            combined_data = np.concatenate([list_values, dict_values])
            self.shm = shared_memory.SharedMemory(create=True, size=combined_data.nbytes)
            self.shared_arr = np.ndarray(combined_data.shape, dtype=combined_data.dtype, buffer=self.shm.buf)
            self.shared_arr[:] = combined_data[:]
            self.shm_name = self.shm.name
        else:
            # 子进程：从共享内存加载数据
            self.shm = shared_memory.SharedMemory(name=shm_name)
            total_len = 1000 + 5000
            self.shared_arr = np.ndarray((total_len,), dtype=np.int64, buffer=self.shm.buf)
            # 拆分回原来的数据集
            self.list_values = self.shared_arr[:1000]
            self.dict_values = self.shared_arr[1000:]

def cpu_task(shm_name, task_num):
    # 子进程初始化对象，从共享内存加载数据
    obj = ComplexObj(shm_name=shm_name)
    total = obj.list_values.sum() + obj.dict_values.sum()
    print(f"Task {task_num} finished, total: {total}")

if __name__ == "__main__":
    # 主进程创建共享对象
    shared_obj = ComplexObj()

    # 启动并行进程
    processes = [mp.Process(target=cpu_task, args=(shared_obj.shm_name, i)) for i in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

    # 主进程负责清理共享内存
    shared_obj.shm.close()
    shared_obj.shm.unlink()

这个方案的关键是避免了整个复杂对象的pickle复制，只共享核心计算数据，跨平台可用，开销也很小。

3. 绝对避坑：别用multiprocessing.Manager #

很多新手会想到用Manager来共享对象，但Manager是通过一个单独的服务器进程来管理所有共享数据，子进程对共享对象的每一次访问都要走IPC通信——对于CPU密集型任务来说，这个通信开销会大到离谱，比复制对象还慢，完全不适合你的场景。

总结 #

• 优先用Unix系统+fork启动方式，利用写时复制，零成本高效
• 必须跨平台就用shared_memory提取核心数据到共享内存
• 打死别碰Manager处理只读复杂对象

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