如何基于现有NumPy数组初始化共享内存，实现子进程无拷贝访问

这个问题我太有共鸣了——GB级的大数组要是复制一次，不仅耗时还占满内存，完全是得不偿失。下面分场景给你最实用的解决方案：

一、Unix系统（Linux/macOS）：利用fork的写时复制特性（零复制） #

Unix下multiprocessing默认用fork启动子进程，这种方式下子进程会共享父进程的内存空间，而且采用**写时复制（COW）**机制：只有当某一方修改内存页时，系统才会复制该页数据，纯读操作完全不会复制整个数组。

直接上代码：

import multiprocessing as mp
import numpy as np

def process_task(arr):
    # 子进程直接访问数组，纯读操作不会触发复制
    print("子进程读取数组前5个元素:", arr[:5])
    # 如果要修改：注意这里会触发对应内存页的复制，但依然不会复制整个数组
    # arr[0] = 999

if __name__ == "__main__":
    # 模拟超大NumPy数组（比如1亿个float64元素，约800MB）
    big_array = np.random.rand(10**8)
    
    # 启动子进程，直接传递数组
    proc = mp.Process(target=process_task, args=(big_array,))
    proc.start()
    proc.join()

这种方法最简单，而且完全零复制，是Unix环境下的最优解。

二、跨平台方案（含Windows）：用共享内存直接映射 #

Windows不支持fork，默认用spawn启动子进程，会序列化传递的对象，导致数组被完整复制。这时候就需要用multiprocessing.shared_memory（Python3.8+新增）来创建共享内存块，让子进程直接映射这块内存。

注意：如果数组已经存在，必须做一次复制到共享内存（唯一的一次开销）；如果可以提前规划，最好直接在共享内存中创建数组，彻底避免复制： #

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

def process_task(shm_name, arr_shape, arr_dtype):
    # 连接到父进程创建的共享内存块
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    # 从共享内存直接构建NumPy数组，无复制
    shared_arr = np.ndarray(arr_shape, dtype=arr_dtype, buffer=existing_shm.buf)
    
    print("子进程读取数组前5个元素:", shared_arr[:5])
    
    # 用完关闭共享内存（不要unlink，父进程负责释放）
    existing_shm.close()

if __name__ == "__main__":
    # 方案1：直接在共享内存中创建数组（零复制）
    arr_shape = (10**8,)
    arr_dtype = np.float64
    # 创建共享内存块，大小匹配数组字节数
    shm = shared_memory.SharedMemory(create=True, size=arr_shape[0] * arr_dtype.itemsize)
    # 基于共享内存构建NumPy数组
    big_array = np.ndarray(arr_shape, dtype=arr_dtype, buffer=shm.buf)
    # 可以正常给数组赋值，数据直接存在共享内存中
    big_array[:] = np.random.rand(*arr_shape)

    # 方案2：如果数组已经存在，需要复制到共享内存（仅一次开销）
    # big_array = np.random.rand(10**8)
    # shm = shared_memory.SharedMemory(create=True, size=big_array.nbytes)
    # shared_arr = np.ndarray(big_array.shape, dtype=big_array.dtype, buffer=shm.buf)
    # shared_arr[:] = big_array[:]  # 这里会复制一次数据

    # 启动子进程，传递共享内存名称、数组形状和类型
    proc = mp.Process(target=process_task, args=(shm.name, arr_shape, arr_dtype))
    proc.start()
    proc.join()

    # 父进程最后关闭并释放共享内存
    shm.close()
    shm.unlink()

三、旧版本Python（3.8以下）：用multiprocessing.Array映射 #

如果你的Python版本低于3.8，可以用multiprocessing.Array创建共享内存数组，再用np.frombuffer将其转换为NumPy数组（无复制）：

import multiprocessing as mp
import numpy as np

def process_task(shared_arr, arr_shape):
    # 从共享内存数组映射为NumPy数组，无复制
    np_arr = np.frombuffer(shared_arr.get_obj(), dtype=np.float64).reshape(arr_shape)
    print("子进程读取数组前5个元素:", np_arr[:5])

if __name__ == "__main__":
    arr_shape = (10**8,)
    # 创建共享内存数组，'d'对应float64类型
    shared_arr = mp.Array('d', arr_shape[0])
    # 转换为NumPy数组并赋值
    np_arr = np.frombuffer(shared_arr.get_obj(), dtype=np.float64).reshape(arr_shape)
    np_arr[:] = np.random.rand(*arr_shape)

    proc = mp.Process(target=process_task, args=(shared_arr, arr_shape))
    proc.start()
    proc.join()

关键提醒 #

• Unix下优先用fork方案，零复制效率最高；
• Windows下无法避免首次复制（除非提前在共享内存创建数组）；
• 如果子进程需要修改数组，必须用共享内存方案（fork的写时复制会导致父子进程数据分离）。

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