如何在原生Python代码中持久化大数组至内存以实现快速复用？

针对你的需求（Python 3.6、Linux环境、大数组常驻内存、客户端可修改代码重启、无pickle开销），multiprocessing的共享数组方案完全可行，同时还有更适合你场景的内存映射（mmap）方案，以下是具体实现：

方案1：基于mmap的内存映射文件（推荐） #

Linux下的内存映射文件会被系统缓存到内存，服务端一次性加载数组后写入映射文件并保持运行，客户端直接映射该文件到自身进程空间，无需重复读取磁盘，也无对象序列化开销。

服务端（常驻进程） #

import pickle
import mmap
import os
import numpy as np

FILE_PATH_TO_PICKLED_ARRAY = "large_array.pkl"
MMAP_FILE_PATH = "large_array_mmap.bin"

def main():
    # 一次性加载磁盘中的大数组
    with open(FILE_PATH_TO_PICKLED_ARRAY, "rb") as f:
        arr = pickle.load(f)
    
    # 导出数组二进制数据及元信息（形状、类型）
    arr_bytes = arr.tobytes()
    arr_meta = (arr.shape, arr.dtype)
    
    # 保存元信息供客户端读取
    with open(f"{MMAP_FILE_PATH}_meta", "wb") as f:
        pickle.dump(arr_meta, f)
    
    # 写入二进制数据到内存映射文件
    with open(MMAP_FILE_PATH, "wb") as f:
        f.write(arr_bytes)
    
    # 映射文件到内存，保持进程运行以维持系统缓存
    with open(MMAP_FILE_PATH, "r+b") as f:
        mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
        print("服务端启动完成，数组已常驻内存。按Ctrl+C终止服务。")
        try:
            while True:
                os.sleep(3600)
        except KeyboardInterrupt:
            mm.close()
            print("服务端已终止。")

if __name__ == "__main__":
    main()

客户端（可频繁修改重启） #

import mmap
import numpy as np
import pickle

MMAP_FILE_PATH = "large_array_mmap.bin"

def some_task(arr):
    # 此处可自由修改实现，重启客户端即可生效
    result = arr.sum()
    print(f"任务执行结果：{result}")
    return result

def main():
    # 读取数组元信息
    with open(f"{MMAP_FILE_PATH}_meta", "rb") as f:
        arr_shape, arr_dtype = pickle.load(f)
    
    # 映射内存文件到当前进程，直接访问内存数据
    with open(MMAP_FILE_PATH, "r+b") as f:
        mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
        # 将内存映射转为numpy数组（无数据拷贝，直接操作共享内存）
        arr = np.frombuffer(mm, dtype=arr_dtype).reshape(arr_shape)
        
        # 执行任务，arr为只读状态
        some_task(arr)
        
        mm.close()

if __name__ == "__main__":
    main()

优势：

• 服务端只需启动一次，后续客户端访问完全走内存，无磁盘IO开销。
• 客户端修改代码后直接重启即可，无需等待数组加载。
• 依赖完全原生，无需额外安装库。

方案2：基于multiprocessing.Array的命名共享内存 #

利用Python原生的multiprocessing模块创建命名共享内存块，不同进程通过名称访问同一内存区域，无对象序列化开销。

服务端（常驻进程） #

import pickle
import multiprocessing
import numpy as np
import os

FILE_PATH_TO_PICKLED_ARRAY = "large_array.pkl"
SHARED_ARRAY_NAME = "my_large_shared_array"

# 映射numpy dtype到multiprocessing对应的ctypes类型
DTYPE_TO_CTYPES = {
    np.float64: multiprocessing.c_double,
    np.float32: multiprocessing.c_float,
    np.int64: multiprocessing.c_longlong,
    np.int32: multiprocessing.c_int,
}

def main():
    # 加载大数组
    with open(FILE_PATH_TO_PICKLED_ARRAY, "rb") as f:
        arr = pickle.load(f)
    
    # 创建命名共享数组
    ctype = DTYPE_TO_CTYPES[arr.dtype.type]
    shared_arr = multiprocessing.Array(ctype, arr.size, name=SHARED_ARRAY_NAME)
    
    # 将数组数据复制到共享内存（仅执行一次）
    np_shared = np.frombuffer(shared_arr.get_obj(), dtype=arr.dtype).reshape(arr.shape)
    np_shared[:] = arr[:]
    
    # 保存数组元信息
    with open(f"{SHARED_ARRAY_NAME}_meta", "wb") as f:
        pickle.dump((arr.shape, arr.dtype), f)
    
    print("服务端启动完成，共享数组已常驻内存。按Ctrl+C终止服务。")
    try:
        while True:
            os.sleep(3600)
    except KeyboardInterrupt:
        print("服务端已终止，共享内存将被释放。")

if __name__ == "__main__":
    main()

客户端（可频繁修改重启） #

import multiprocessing
import numpy as np
import pickle

SHARED_ARRAY_NAME = "my_large_shared_array"

def some_task(arr):
    # 此处可自由修改实现，重启客户端即可生效
    result = arr.mean()
    print(f"任务执行结果：{result}")
    return result

def main():
    # 读取数组元信息
    with open(f"{SHARED_ARRAY_NAME}_meta", "rb") as f:
        arr_shape, arr_dtype = pickle.load(f)
    
    # 连接到已有的命名共享数组
    shared_arr = multiprocessing.Array(None, name=SHARED_ARRAY_NAME)
    
    # 转为numpy数组（无数据拷贝，直接操作共享内存）
    arr = np.frombuffer(shared_arr.get_obj(), dtype=arr.dtype).reshape(arr_shape)
    
    # 执行任务
    some_task(arr)

if __name__ == "__main__":
    main()

注意：

• Python 3.6中该方案基于POSIX共享内存实现，服务端终止后共享内存会被自动清理。
• 若数组类型不在映射表中，需自行补充对应的ctypes类型。

通用注意事项 #

• 两种方案均避免了对象pickle的开销（仅极小的元信息使用pickle）。
• 若你使用的是普通Python列表而非numpy数组，需先将列表转为二进制格式（如用struct模块），但numpy数组的处理效率更高，推荐优先使用。

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