如何并行化写入全局变量的函数？基于多核集群的实现

核心问题说明 #

Python多进程间无法直接共享普通全局变量，直接使用你代码里的data_store1/data_store2会导致每个进程拥有独立副本，修改不会同步到主进程。因此必须采用共享内存来存储需要跨进程修改的数组，同时结合进程池最大化利用128核集群的资源。

具体实现方案 #

1. 基于multiprocessing的共享内存+进程池方案

利用numpy数组的连续内存特性，将存储数组转为共享内存对象，让所有进程直接操作同一块内存区域，避免数据拷贝开销。

import multiprocessing as mp
import numpy as np

def create_shared_array(shape, dtype):
    # 创建共享内存块并转为numpy数组
    shared_mem = mp.Array(np.dtype(dtype).char, shape[0])
    return np.frombuffer(shared_mem.get_obj(), dtype=dtype).reshape(shape)

def process_task(args):
    idx, val = args
    # 替换为你simulation函数中的实际逻辑：从全局数据取值并设置对应存储元素
    data_store1[idx] = val * global_data2[idx]
    data_store2[idx] = val + global_data2[idx]

if __name__ == "__main__":
    n = 100000  # 替换为你的实际数据长度
    global_data = np.random.rand(n)  # 替换为你的实际全局数据
    global_data2 = np.random.rand(n)

    # 创建共享内存的存储数组
    data_store1 = create_shared_array((n,), np.float64)
    data_store2 = create_shared_array((n,), np.float64)

    # 将大体积全局数据转为共享内存，避免进程间拷贝浪费资源
    global_data_shared = create_shared_array((n,), np.float64)
    global_data_shared[:] = global_data[:]
    global_data2_shared = create_shared_array((n,), np.float64)
    global_data2_shared[:] = global_data2[:]

    # 进程池匹配集群核心数，自动分配任务
    with mp.Pool(processes=mp.cpu_count()) as pool:
        pool.map(process_task, enumerate(global_data_shared))

    # 任务完成后直接使用data_store1/2即可
    print("处理后的data_store1前10个元素：", data_store1[:10])

2. 关键细节说明

• 共享内存的必要性：普通全局变量在多进程中是独立拷贝，只有通过mp.Array创建的对象才能被所有进程共享修改。
• 进程池优势：自动管理进程生命周期，实现负载均衡，避免手动创建128个进程的额外开销，mp.cpu_count()会自动匹配集群核心数。
• 无锁操作：每个进程仅操作数组的独立索引，不存在资源竞争，无需加锁，性能最优。

3. 替代方案：concurrent.futures.ProcessPoolExecutor

语法更简洁，功能与mp.Pool一致，适合习惯现代Python API的场景：

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    # 初始化共享数组和数据逻辑同上...
    with ProcessPoolExecutor(max_workers=mp.cpu_count()) as executor:
        executor.map(process_task, enumerate(global_data_shared))

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