Python 并发利器：multiprocessing 共享内存 SharedMemory 零拷贝实录

Python 并发利器：multiprocessing 共享内存 SharedMemory 零拷贝实录

在 Python 中进行多进程并行计算（如特征预处理、大规模图像渲染或模型多进程推理）时，开发者最常使用的通信工具是 multiprocessing.Queue 或 Pipe。

然而，当进程间需要传递的是大尺寸的 NumPy 数组（例如一个 500MB 的特征矩阵或高分辨率视频帧序列）时，传统的队列通信机制会成为可怕的性能黑洞。

其根本原因在于：Queue 和 Pipe 在底层必须对数据进行 Pickle 序列化与跨进程 IPC 管道拷贝。传递一次 500MB 数据，不仅会在主进程和子进程中各复制一份（导致内存占用瞬间翻倍），还会将 CPU 核心完全耗尽在繁重的内存搬运上。

Python 3.8+ 引入的 multiprocessing.shared_memory.SharedMemory 提供了真正的系统级共享内存支持，能够实现多进程间零拷贝（Zero-copy）、纳秒级数据共享。

1. 传统 IPC 拷贝与 SharedMemory 的底层机制对比

• 传统 multiprocessing.Queue 路径：进程 A 内存 $\\xrightarrow{\\text{Pickle 序列化}}$ 临时内存 Buffer $\\xrightarrow{\\text{write() 系统调用}}$ 操作系统管道/Socket $\\xrightarrow{\\text{read() 系统调用}}$ 临时内存 Buffer $\\xrightarrow{\\text{Unpickle 反序列化}}$ 进程 B 内存（产生至少 3 次全量内存拷贝与数百万次内存分配）；
• SharedMemory 路径：在操作系统内核（如 Linux 的 /dev/shm POSIX 共享内存段）中开辟一块物理内存，进程 A 和进程 B 通过 mmap 将其直接映射到各自的虚拟地址空间。两进程直接读写同一块物理内存指针，内存拷贝次数恒等于 0。

2. 基于 SharedMemory 与 NumPy 的零拷贝并发封装

以下是一个工业级可复用的共享内存管理脚手架：

import numpy as np
from multiprocessing import Process, shared_memory
import time
from typing import Tuple

def worker_process(shm_name: str, shape: Tuple[int, …], dtype: np.dtype, worker_id: int, total_workers: int):
# 1. 在子进程中连接现有的共享内存块
existing_shm = shared_memory.SharedMemory(name=shm_name)

# 2. 基于共享内存缓冲区零拷贝构建 NumPy 数组视图
shared_array = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)

# 3. 按切片分配各 Worker 计算任务 (原地原地修改内存)
chunk_size = shape[0] // total_workers
start = worker_id * chunk_size
end = shape[0] if worker_id == total_workers – 1 else (worker_id + 1) * chunk_size

# 执行复杂数值计算 (例如求正弦变换)
shared_array[start:end] = np.sin(shared_array[start:end]) + np.cos(shared_array[start:end])

# 4. 关闭子进程视图描述符
existing_shm.close()

def run_shared_memory_pipeline():
# 创建一个 10,000,000 个 float64 元素的庞大矩阵 (约 80MB)
N = 10_000_000
dtype = np.float64
shape = (N,)

# 1. 在主进程中分配 POSIX 共享内存
shm = shared_memory.SharedMemory(create=True, size=int(np.prod(shape) * np.dtype(dtype).itemsize))

# 将 NumPy 数组挂载到共享内存上并填充初始值
main_array = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
main_array[:] = np.linspace(0, 100, N)

num_workers = 4
processes = []

t0 = time.perf_counter()
# 2. 启动子进程，仅需传递共享内存名称字符串，零数据序列化开销
for i in range(num_workers):
p = Process(target=worker_process, args=(shm.name, shape, dtype, i, num_workers))
processes.append(p)
p.start()

for p in processes:
p.join()
t1 = time.perf_counter()

print(f"4 进程共享内存并行计算耗时: {(t1 – t0) * 1000:.2f} ms")

# 3. 释放并销毁共享内存
shm.close()
shm.unlink() # 彻底归还操作系统内存

3. 性能基准压测对比

我们测试 4 个子进程并发读取并处理一个 500MB 大小的特征矩阵：

通信开销由近 2 秒缩短至 12 毫秒，提速超过 150 倍，且彻底消除了多进程同时拷贝造成的内存 OOM 隐患。

4. 关键避坑与资源释放守则