本文详解如何在 Python 多进程中安全、高效地共享并原地修改同一个 NumPy 数组，解决 joblib 和默认 multiprocessing.Pool 因进程隔离导致的内存不共享问题。核心方案是借助 multiprocessing.Array 构建共享内存缓冲区，并通过 np.ndarray(..., buffer=...) 重建可写视图。
本文详解如何在 python 多进程中安全、高效地共享并原地修改同一个 numpy 数组，解决 `joblib` 和默认 `multiprocessing.pool` 因进程隔离导致的内存不共享问题。核心方案是借助 `multiprocessing.array` 构建共享内存缓冲区，并通过 `np.ndarray(..., buffer=...)` 重建可写视图。
在 Python 多进程编程中，子进程默认拥有父进程数据的独立副本（通过 fork 或 spawn 实现），因此直接将 NumPy 数组传入 Pool.map 或 joblib.Parallel 后对数组的修改不会反映到主进程中——这正是原始代码失败的根本原因。要实现“并行修改同一数组”，必须显式使用共享内存（shared memory），而非传递副本。
关键在于：NumPy 数组本质是带元信息（shape/dtype/strides）的内存视图。只要让所有进程操作同一块底层内存（如 multiprocessing.Array），再为每个进程绑定对应的 NumPy 视图，即可实现真正的并发写入。
以下为完整可运行方案：
import numpy as np
import multiprocessing
def np_array_to_shared_array(np_array: np.ndarray, lock: bool = True) -> multiprocessing.Array:
"""将 NumPy 数组内容复制到 multiprocessing.Array 共享内存中"""
shared_array = multiprocessing.Array('B', np_array.nbytes, lock=lock)
buffer = shared_array.get_obj() if lock else shared_array
arr = np.frombuffer(buffer, dtype=np_array.dtype)
arr[:] = np_array.flatten(order='C') # 按 C 顺序展平写入
return shared_array
def shared_array_to_np_array(shared_array: multiprocessing.Array,
shape: tuple, dtype: np.dtype) -> np.ndarray:
"""从共享内存构建 NumPy 数组视图（不拷贝数据）"""
buffer = shared_array.get_obj() if hasattr(shared_array, 'get_obj') else shared_array
return np.ndarray(shape, dtype=dtype, buffer=buffer)
# 全局变量，供子进程访问共享数组
a = None
def init_pool(shared_array, shape, dtype):
"""池初始化函数：为每个子进程设置全局共享数组视图"""
global a
a = shared_array_to_np_array(shared_array, shape, dtype)
def test(i):
"""业务逻辑：根据 i 的奇偶性修改数组偶/奇索引位置"""
n = a.shape[0]
if i % 2 == 0:
for x in range(0, n, 2):
a[x] = 2
else:
for x in range(1, n, 2):
a[x] = 1
def main():
# 初始化主进程数组
arr = np.zeros(10, dtype=np.uint8)
shape, dtype = arr.shape, arr.dtype
# 创建共享内存并初始化
shared_array = np_array_to_shared_array(arr, lock=False)
# 在主进程中也映射为 NumPy 视图（可选，用于验证）
arr_mapped = shared_array_to_np_array(shared_array, shape, dtype)
# 启动进程池，指定初始化函数和参数
with multiprocessing.Pool(
processes=2,
initializer=init_pool,
initargs=(shared_array, shape, dtype)
) as pool:
pool.map(test, [0, 1]) # 并行执行两个任务
print(arr_mapped) # 输出: [2 1 2 1 2 1 2 1 2 1]
if __name__ == '__main__':
main()
✅ 关键要点说明：
-
multiprocessing.Array('B', size)中'B'表示无符号字节（uint8），需与 NumPydtype对齐；若使用int32等类型，应改用'i'并调整nbytes计算方式（推荐统一用ctypes类型如ctypes.c_uint8提升可读性）。 -
lock=False可提升性能，但要求业务逻辑严格避免竞态写入同一内存位置；本例中i=0写偶数索引、i=1写奇数索引，天然无冲突，故可安全关闭锁。 -
init_pool通过initializer参数在每个子进程启动时执行一次，确保a在各进程内均为指向共享内存的有效 NumPy 视图。 - 主进程中的
arr_mapped与子进程操作的是同一内存块，因此打印结果即为并行修改后的最终状态。
⚠️ 注意事项：
- 此方案不适用于 Windows 上的
spawn启动方式（默认），需显式设mp.set_start_method('fork')（仅 Linux/macOS 支持）；若需跨平台兼容，建议改用multiprocessing.shared_memory（Python 3.8+）或ray/dask等高级框架。 - 避免在
test函数中重新分配a（如a = ...），否则会切断与共享内存的绑定。 - 大数组场景下，
np_array_to_shared_array的初始化拷贝不可避免，但后续所有写操作均零拷贝。
该方法平衡了性能、可控性与可维护性，是科学计算中多进程原地修改大型数组的经典实践。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
