如何高效处理大型自定义数据结构对象列表？多进程内存优化咨询

我需要读取并修改由相同大型自定义数据结构（包含numpy.array、list、dict等）组成的列表中的元素。自定义数据结构定义如下：

class DataStruct:
    def __init__(self, a, b, c, others) -> None:
        self.a = a # numpy.array
        self.b = b # numpy array
        self.c = c # numpy array
        self.others = others

    def update_a(self, value):
        pass

    def update_b(self, value):
        pass

    def update_c(self, value):
        pass

    def update_others(self, values):
        pass

    def run(self):
        # run algorithm by processing (a, b, c, others)
        pass

我尝试使用multiprocessing包为每个对象分配一个进程处理，但性能远低于串行处理——原因在于进程间传递数据需要复制，且进程启动存在开销。我已尝试使用multiprocessing.Queue、Manager以及Ray包，但均未解决问题：Ray仅允许复制对象到本地后修改，无法直接共享修改。

请问是否有方法实现进程间共享这类数据结构对象的内存，或避免数据复制？

可行解决方案 #

1. 使用multiprocessing共享内存数组 + NumPy视图 #

利用multiprocessing.Array或RawArray创建共享内存块，再将其转换为NumPy数组视图，所有进程可直接访问同一块内存，无需复制数据。同时需用锁保证多进程修改的线程安全。

修改DataStruct类以支持共享内存：

import multiprocessing as mp
import numpy as np

class SharedDataStruct:
    def __init__(self, a_shape, b_shape, c_shape, others, lock=None):
        # 创建共享内存数组，指定数据类型和形状
        self.a_shared = mp.Array('d', int(np.prod(a_shape)))
        self.b_shared = mp.Array('d', int(np.prod(b_shape)))
        self.c_shared = mp.Array('d', int(np.prod(c_shape)))
        
        # 将共享内存转换为NumPy数组视图
        self.a = np.frombuffer(self.a_shared.get_obj(), dtype='d').reshape(a_shape)
        self.b = np.frombuffer(self.b_shared.get_obj(), dtype='d').reshape(b_shape)
        self.c = np.frombuffer(self.c_shared.get_obj(), dtype='d').reshape(c_shape)
        
        self.others = others
        self.lock = lock or mp.Lock()

    def update_a(self, value):
        with self.lock:
            self.a[:] = value

    def run(self):
        with self.lock:
            # 直接操作共享内存中的数组
            self.a += self.b * self.c

使用方式：

def process_struct(struct):
    struct.run()

if __name__ == "__main__":
    # 初始化共享内存数据结构
    struct = SharedDataStruct(
        a_shape=(1000, 1000), 
        b_shape=(1000, 1000), 
        c_shape=(1000, 1000),
        others={"key": "value"}
    )
    
    # 填充初始数据
    struct.a[:] = np.random.rand(1000, 1000)
    struct.b[:] = np.random.rand(1000, 1000)
    struct.c[:] = np.random.rand(1000, 1000)
    
    # 启动多进程处理
    processes = [mp.Process(target=process_struct, args=(struct,)) for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

2. 使用multiprocessing.shared_memory（Python 3.8+） #

Python 3.8引入的shared_memory模块更简洁，专门用于跨进程共享内存块。创建SharedMemory对象后，NumPy数组可直接基于该内存构建，进程间只需传递共享内存名称即可。

示例代码：

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

class SharedDataStruct:
    def __init__(self, a=None, b=None, c=None, others=None, shm_names=None):
        if shm_names:
            # 从已有的共享内存加载
            a_shm = shared_memory.SharedMemory(name=shm_names['a'])
            b_shm = shared_memory.SharedMemory(name=shm_names['b'])
            c_shm = shared_memory.SharedMemory(name=shm_names['c'])
            
            self.a = np.ndarray(a.shape, dtype=a.dtype, buffer=a_shm.buf)
            self.b = np.ndarray(b.shape, dtype=b.dtype, buffer=b_shm.buf)
            self.c = np.ndarray(c.shape, dtype=c.dtype, buffer=c_shm.buf)
            
            self._shms = [a_shm, b_shm, c_shm]
        else:
            # 创建新的共享内存
            a_shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
            b_shm = shared_memory.SharedMemory(create=True, size=b.nbytes)
            c_shm = shared_memory.SharedMemory(create=True, size=c.nbytes)
            
            self.a = np.ndarray(a.shape, dtype=a.dtype, buffer=a_shm.buf)
            self.b = np.ndarray(b.shape, dtype=b.dtype, buffer=b_shm.buf)
            self.c = np.ndarray(c.shape, dtype=c.dtype, buffer=c_shm.buf)
            
            # 复制初始数据到共享内存
            self.a[:] = a
            self.b[:] = b
            self.c[:] = c
            
            self._shms = [a_shm, b_shm, c_shm]
            self.shm_names = {'a': a_shm.name, 'b': b_shm.name, 'c': c_shm.name}
        
        self.others = others
        self.lock = mp.Lock()

    def close(self):
        for shm in self._shms:
            shm.close()
    
    def unlink(self):
        for shm in self._shms:
            shm.unlink()

def process_task(shm_names, a_shape, a_dtype, b_shape, b_dtype, c_shape, c_dtype, others):
    # 子进程中加载共享内存数据结构
    a = np.ndarray(a_shape, dtype=a_dtype)
    b = np.ndarray(b_shape, dtype=b_dtype)
    c = np.ndarray(c_shape, dtype=c_dtype)
    struct = SharedDataStruct(a=a, b=b, c=c, others=others, shm_names=shm_names)
    
    with struct.lock:
        struct.run()
    
    struct.close()

if __name__ == "__main__":
    # 初始化原始数据
    a = np.random.rand(1000, 1000)
    b = np.random.rand(1000, 1000)
    c = np.random.rand(1000, 1000)
    others = {"key": "value"}
    
    # 创建共享内存数据结构
    struct = SharedDataStruct(a=a, b=b, c=c, others=others)
    
    # 启动子进程，传递共享内存名称和数组元信息
    p = mp.Process(
        target=process_task,
        args=(
            struct.shm_names,
            a.shape, a.dtype,
            b.shape, b.dtype,
            c.shape, c.dtype,
            others
        )
    )
    p.start()
    p.join()
    
    # 清理共享内存
    struct.unlink()

3. 使用线程而非进程（适用于GIL不阻塞的场景） #

如果run方法主要是NumPy运算（NumPy内部会释放GIL，支持多线程并行）或IO密集型操作，使用线程可完全避免进程间的数据复制开销——线程共享同一进程的内存空间。

示例代码：

import threading
import numpy as np

class DataStruct:
    def __init__(self, a, b, c, others) -> None:
        self.a = a # numpy.array
        self.b = b # numpy array
        self.c = c # numpy array
        self.others = others
        self.lock = threading.Lock()

    def update_a(self, value):
        with self.lock:
            self.a[:] = value

    def run(self):
        with self.lock:
            self.a += self.b * self.c

def process_struct(struct):
    struct.run()

if __name__ == "__main__":
    # 初始化数据结构列表
    structs = []
    for _ in range(4):
        a = np.random.rand(1000, 1000)
        b = np.random.rand(1000, 1000)
        c = np.random.rand(1000, 1000)
        structs.append(DataStruct(a, b, c, {"key": "value"}))
    
    # 启动多线程处理
    threads = [threading.Thread(target=process_struct, args=(s,)) for s in structs]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

4. 进程池+初始化器传递共享数据 #

使用multiprocessing.Pool的initializer参数，在子进程启动时一次性加载共享数据，避免每次任务传递数据的复制开销。

示例：

import multiprocessing as mp
import numpy as np

shared_struct = None
lock = None

def init_worker(struct, worker_lock):
    global shared_struct, lock
    shared_struct = struct
    lock = worker_lock

def process_task(index):
    with lock:
        # 处理指定逻辑（如果是列表可通过索引区分）
        shared_struct.run()

if __name__ == "__main__":
    # 初始化共享内存数据结构
    struct = SharedDataStruct(
        a_shape=(1000, 1000), 
        b_shape=(1000, 1000), 
        c_shape=(1000, 1000),
        others={"key": "value"}
    )
    struct.a[:] = np.random.rand(1000, 1000)
    struct.b[:] = np.random.rand(1000, 1000)
    struct.c[:] = np.random.rand(1000, 1000)
    
    lock = mp.Lock()
    
    # 创建进程池，传递共享结构和锁
    with mp.Pool(initializer=init_worker, initargs=(struct, lock)) as pool:
        pool.map(process_task, range(4))

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