Python Shared Memory 的代价：多个 Worker 如何高速、零拷贝且安全地共享大型 NumPy Array

Python Shared Memory 的代价：多个 Worker 如何高速、零拷贝且安全地共享大型 NumPy Array

在 Python 性能优化中，有一个非常诱人的设想：

我有一个几 GB 的 NumPy 数组，希望 8 个、16 个甚至更多 worker 同时处理它。能不能只保存一份数据，不经过 pickle，不复制，同时又跑得足够快？

答案是：可以，但 Shared Memory 从来不是免费的午餐。

它解决了“数据复制”的问题，却同时把 ownership、生命周期、同步、缓存一致性、故障恢复以及内存带宽竞争 全部推到了开发者面前。

如果只是写一个 Demo，multiprocessing.shared_memory 十几行代码就够了；但如果要放进长期运行的 Python 服务、数据处理平台或者实时计算系统，真正困难的问题往往是：

• 谁创建这块内存，谁拥有它？
• 哪个进程有资格执行 unlink()？
• worker 崩溃以后谁清理？
• 多个 worker 同时修改数组，是否真的安全？
• 明明不同 worker 写的是不同元素，为什么性能反而下降？
• 8 个进程一定比 4 个进程快吗？
• 所谓“零拷贝”究竟有没有真的做到？

本文就从这些问题出发，拆解 Python Shared Memory 在大型 NumPy Array 场景中的正确使用方式。

一、Shared Memory 到底解决了什么问题？

普通 multiprocessing 最大的隐藏成本之一，是跨进程数据传输。

假设有一个：

import numpy as np

arr = np.random.random((100_000_000,))

如果使用 float64，它大约占：

100,000,000 × 8 bytes
≈ 800 MB

如果把它直接传入 ProcessPoolExecutor：

executor.submit(worker, arr)

通常意味着需要经历：

NumPy Array
↓
pickle / serialization
↓
IPC
↓
deserialize
↓
worker 私有内存

假如启动 8 个 worker，最坏情况下不仅会产生大量序列化开销，还可能出现多份数组副本。

Shared Memory 的思路完全不同：

Shared Memory
┌────────────────┐
│ Large ndarray │
└────────────────┘
↑ ↑ ↑
│ │ │
Worker1 Worker2 Worker3

不同进程建立自己的 ndarray 对象，但底层数据指向同一段共享内存。

Python 官方的 multiprocessing.shared_memory 就是为这种跨进程共享内存场景设计的；NumPy 本身也允许通过已有 buffer 创建 ndarray，因此二者天然可以组合。(Python documentation)

但这里有第一个非常重要的认识：

Shared Memory 实现的是“共享底层数据”，不是共享 Python 对象。

每个 worker 都有自己的：

ndarray object
shape
strides
dtype metadata

真正共享的是：

raw memory buffer

因此：

np.ndarray(
shape,
dtype=dtype,
buffer=shm.buf,
)

实际上是在告诉 NumPy：

不要分配新的数据区，直接把这块 buffer 当作数组数据。

这正是零拷贝访问的基础。

二、先搞清一个概念：你所谓的“零拷贝”可能并不是真正零拷贝

很多 Shared Memory 示例都会这样写：

shm = shared_memory.SharedMemory(
create=True,
size=arr.nbytes,
)

shared_arr = np.ndarray(
arr.shape,
dtype=arr.dtype,
buffer=shm.buf,
)

shared_arr[:] = arr

注意最后一行：

shared_arr[:] = arr

这里发生了一次完整的数据复制。

所以更准确地说，它实现的是：

初始化：
普通 ndarray
↓ copy
Shared Memory

之后：
worker → Shared Memory
zero-copy access

也就是说，worker 访问阶段零拷贝，并不代表整个数据生命周期零拷贝。

真正希望从头避免一次大数组复制，应让生产者直接写入 Shared Memory：

from multiprocessing import shared_memory
import numpy as np

shape = (100_000_000,)
dtype = np.float64

shm = shared_memory.SharedMemory(
create=True,
size=np.prod(shape) * np.dtype(dtype).itemsize,
)

arr = np.ndarray(
shape,
dtype=dtype,
buffer=shm.buf,
)

# 数据直接产生在 shared memory 中
arr[:] = np.random.random(shape)

当然，随机数生成过程中仍然存在 CPU 和内存写入，但已经不需要再把一个完整 ndarray 搬运到另一块共享区域。

三、一个可用于实际项目的 Shared NumPy Array

先实现最核心的版本。

import numpy as np
from multiprocessing import shared_memory

def create_shared_array(shape, dtype=np.float64):
dtype = np.dtype(dtype)

nbytes = (
int(np.prod(shape))
* dtype.itemsize
)

shm = shared_memory.SharedMemory(
create=True,
size=nbytes,
)

arr = np.ndarray(
shape,
dtype=dtype,
buffer=shm.buf,
)

return shm, arr

创建：

shm, arr = create_shared_array(
(20_000_000,),
np.float64,
)

arr[:] = np.arange(arr.size)

worker 不需要接收数组本体，只需要接收几个很小的元数据：

shared memory name
shape
dtype
start
end

例如：

from multiprocessing import shared_memory
import numpy as np

def worker(name, shape, dtype_str, start, end):
shm = shared_memory.SharedMemory(
name=name,
)

try:
arr = np.ndarray(
shape,
dtype=np.dtype(dtype_str),
buffer=shm.buf,
)

arr[start:end] *= 2

finally:
shm.close()

这里没有把 160MB、800MB 或 10GB 数组传给 worker。

传递的可能只是：

(
"psm_xxxxx",
(20_000_000,),
"<f8",
0,
5_000_000,
)

这正是 Shared Memory 最大的价值。

四、高速 + 不复制 + 安全：关键不是加锁，而是设计 ownership

很多开发者看到“多个进程写共享内存”，第一反应是：

lock = multiprocessing.Lock()

然后：

with lock:
arr[...] = ...

这样当然安全。

但也可能直接把并行程序写回串行程序。

真正高性能的设计原则应该是：

优先通过 ownership 消除竞争，而不是依赖锁解决竞争。

比如有 4 个 worker：

Shared Array

0%────────25%────────50%────────75%────────100%
│ Worker 0 │ Worker 1 │ Worker 2 │ Worker 3 │

代码可以这样分区：

def split_ranges(size, workers):
chunk = (size + workers – 1) // workers

for i in range(workers):
start = i * chunk
end = min(start + chunk, size)

if start < end:
yield start, end

然后每个 worker 只拥有自己的区域：

arr[start:end] *= 2

如果满足：

Worker A 永远不写 Worker B 的区域

那么普通的数据处理阶段根本不需要全局锁。

这可以理解成：

shared storage
+
exclusive write ownership

它通常比：

shared storage
+
global lock

快得多。

五、读共享、写分区，是最推荐的模型

Shared NumPy Array 的几个并发模型里，稳定程度大致可以这样理解：

NumPy 官方也明确提醒：共享数组发生并发读写时需要非常谨慎；多个执行单元同时修改共享数组可能产生 race condition。很多 NumPy 底层计算还会释放 GIL，因此不能指望 GIL 自动保护数组。(NumPy)

换句话说：

arr[i] += 1

不要想当然地认为它是跨线程或跨进程原子的。

这个表达式逻辑上可能包含：

read
↓
add
↓
write

两个 worker 同时操作同一位置，就可能发生 lost update。

六、完整案例：4 个进程零拷贝处理大型数组

下面给一个更接近实际工程的例子。

import multiprocessing as mp
from multiprocessing import shared_memory

import numpy as np

def process_chunk(
shm_name,
shape,
dtype_str,
start,
end,
):
shm = shared_memory.SharedMemory(
name=shm_name,
)

try:
arr = np.ndarray(
shape,
dtype=np.dtype(dtype_str),
buffer=shm.buf,
)

chunk = arr[start:end]

# 每个 worker 独占自己的区域
chunk *= 2.0
np.sqrt(chunk, out=chunk)

finally:
shm.close()

def main():
shape = (20_000_000,)
dtype = np.float64
workers = 4

dtype_obj = np.dtype(dtype)

shm = shared_memory.SharedMemory(
create=True,
size=np.prod(shape) * dtype_obj.itemsize,
)

try:
arr = np.ndarray(
shape,
dtype=dtype_obj,
buffer=shm.buf,
)

# 最理想情况：生产者直接填充共享区域
arr[:] = np.arange(
shape[0],
dtype=dtype,
)

ctx = mp.get_context("spawn")

chunk_size = (
shape[0] + workers – 1
) // workers

processes = []

for worker_id in range(workers):

start = worker_id * chunk_size
end = min(
start + chunk_size,
shape[0],
)

if start >= end:
continue

p = ctx.Process(
target=process_chunk,
args=(
shm.name,
shape,
dtype_obj.str,
start,
end,
),
)

p.start()
processes.append(p)

for p in processes:
p.join()

if p.exitcode != 0:
raise RuntimeError(
f"worker failed: {p.exitcode}"
)

print(arr[:10])

finally:
shm.close()
shm.unlink()

if __name__ == "__main__":
main()

这里有几个值得注意的设计点。

第一，只传：

shm.name
shape
dtype
slice range

不传真正的大数组。

第二，每个 worker：

close()

但不负责：

unlink()

第三，创建 Shared Memory 的 owner 最终统一：

shm.unlink()

这就引出了最关键的生命周期问题。

七、到底谁负责 ownership？

我的工程建议很明确：

谁创建，谁拥有；谁拥有，谁最终负责 unlink。

可以建立这样的角色：

Owner Process
│
├── create SharedMemory
│
├── publish metadata
│
├── launch workers
│
├── monitor workers
│
└── close + unlink
↑
only owner

worker：

attach
↓
use
↓
close

不要让所有 worker 都做：

shm.unlink()

Python 官方文档也明确区分了二者：close() 是关闭当前进程持有的 handle，而 unlink() 是删除底层共享内存块，通常每块共享内存只应该执行一次 unlink()。不同平台的底层实现细节也存在区别。(Python documentation)

因此可以把生命周期规则总结成一句话：

Worker owns handle.
Owner owns resource.

这是非常重要的边界。

八、Owner 崩了怎么办？Crash Cleanup 才是生产环境真正困难的地方

正常退出很好解决：

try:
...
finally:
shm.close()
shm.unlink()

真正麻烦的是：

SIGKILL
OOM Killer
机器掉电
父进程 crash
容器异常退出

这时 finally 根本未必有机会执行。

现代 Python 的 SharedMemory 提供资源跟踪机制；较新的 Python 版本还提供 track 参数，用于让 resource tracker 协助管理共享内存生命周期。对于通过同一个 multiprocessing 祖先进程创建的进程，通常可以共享同一个 tracker。(Python documentation)

但这里还有一个高级坑。

如果是：

Service A
Service B
Service C

三个完全独立启动的 Python 进程，而不是同一个 multiprocessing 进程树，它们可能拥有各自的 resource tracker。

官方文档特别提醒：独立 Python 进程都启用 tracking 时，第一个退出的 tracker 可能提前删除共享内存。这种架构应该明确指定谁承担 bookkeeping，而不是简单让所有独立进程都认为自己负责资源生命周期。(Python documentation)

对于长期运行的服务，我更推荐进一步增加：

registry
+
owner pid
+
creation timestamp
+
heartbeat

例如保存：

{
"name": "my_app_178921",
"owner_pid": 12345,
"created_at": 1789000000,
"shape": [10000000],
"dtype": "<f8",
}

服务启动时执行 stale-resource scan：

发现共享内存
↓
检查 owner PID
↓
owner 是否仍存活？
↓
NO
↓
确认没有合法消费者
↓
cleanup

这比“祈祷 finally 一定执行”可靠得多。

九、SharedMemoryManager：让生命周期简单一些

如果程序结构允许，可以考虑：

from multiprocessing.managers import SharedMemoryManager

with SharedMemoryManager() as smm:

shm = smm.SharedMemory(
size=1024 * 1024 * 1024
)

...

SharedMemoryManager 会专门启动一个管理进程维护它创建的 Shared Memory；manager shutdown 时会清理相应资源。(Python documentation)

它很适合：

batch job
ETL
一次性数据分析任务
父进程明确控制 worker 生命周期

而如果你的系统是：

独立 daemon
多个服务
跨生命周期消费者
共享内存长期存在

那么显式 ownership 通常更容易推理。

十、False Sharing：没有数据竞争，也可能性能暴跌

这是 Shared Memory 最容易被忽略的问题之一。

假设两个 worker：

Worker A → arr[0]
Worker B → arr[1]

它们明明操作不同元素。

是不是完全没有竞争？

逻辑层面没有，CPU Cache 层面未必。

CPU 并不是每次只同步一个 float64。

缓存一致性通常围绕一个 cache line 工作。

于是可能出现：

Cache Line
┌────────────────────────────────┐
│ A0 │ A1 │ A2 │ A3 │ A4 │ … │
└────────────────────────────────┘
↑ ↑
W0 W1

Worker 0 修改 A0：

其他核心对应 cache line 失效

Worker 1 再修改 A1：

cache line ownership 又被抢走

虽然两个进程从来没有修改相同元素，却可能产生持续的：

cache invalidation
cache-line bouncing

这就是 false sharing。

错误分区方式：

# worker 0
arr[0::4]

# worker 1
arr[1::4]

# worker 2
arr[2::4]

# worker 3
arr[3::4]

看起来负载十分平均，但多个 worker 会频繁访问相邻内存。

更好的方式通常是：

# worker 0
arr[0:250000]

# worker 1
arr[250000:500000]

# worker 2
arr[500000:750000]

# worker 3
arr[750000:1000000]

也就是：

优先使用连续的大块 chunk，而不是 interleaved partition。

这样不仅降低 false sharing，还能改善：

cache locality
prefetch
sequential memory access

十一、还有一个更大的敌人：Memory Bandwidth Contention

假设 worker 做的事情非常简单：

arr *= 2

这类运算计算量并不大。

CPU 可能根本不是瓶颈。

真正瓶颈可能是：

RAM
│
Memory Controller
│
CPU Cores
├── Core 1
├── Core 2
├── Core 3
├── …
└── Core 32

当 4 个 worker 已经把内存带宽跑满后：

4 workers → 80 GB/s
8 workers → 81 GB/s
16 workers → 79 GB/s

继续增加进程不仅不会加速，甚至可能因为：

context switching
cache contention
NUMA traffic
scheduler overhead

变得更慢。

所以优化 Shared Memory 程序千万不要简单问：

CPU 有多少核？

更应该问：

这个 workload 是 compute-bound
还是 memory-bound？

如果每读取 8 字节只做一次乘法，它通常很容易成为 memory-bound。

如果读取 8 字节之后进行大量数值计算，并行扩展性才可能明显更好。

十二、生产实践中，我更推荐四条原则

第一，读共享、写私有。

例如模型参数、大型 lookup table、静态特征：

arr.flags.writeable = False

需要注意，writeable=False 更多是一种本进程 NumPy 层面的防误写约束，并不是不可突破的操作系统级安全边界。

第二，必须写的时候使用：

partition ownership

而不是：

everyone writes everywhere

第三，少共享“状态”，多共享“数据”。

不要把 Shared Memory 设计成一个复杂数据库：

state
counter
queue
flag
metadata
payload
lock
version

越复杂，越容易出现 race condition。

控制面适合：

Queue
Pipe
Event
Semaphore
Lock

数据面才适合大型 Shared Memory。

因此很成熟的一种设计其实是：

Control Plane
multiprocessing.Queue
Event
Lock
│
▼
Data Plane
SharedMemory + ndarray

第四，不要使用：

dtype=object

去做这种低层共享。

object 数组中的元素本质上涉及 Python 对象引用，它与纯：

float32
float64
int32
int64

完全不是一回事。

NumPy 对共享可变数组和 object dtype 的线程安全本身也要求额外谨慎。(NumPy)

十三、Shared Memory 是否值得用？先做这组性能测试

不要只测：

1 worker
vs
8 workers

更有价值的是测试：

方案 A：普通 multiprocessing + pickle
方案 B：SharedMemory + 1 worker
方案 C：SharedMemory + 2 workers
方案 D：SharedMemory + 4 workers
方案 E：SharedMemory + 8 workers
方案 F：SharedMemory + 16 workers

记录：

wall-clock latency
throughput
CPU utilization
RSS
memory bandwidth
worker startup time
P95 / P99 latency

你很可能得到类似趋势：

Throughput
^
|
| _________
| __/
| __/
| __/
|___/
+————————–> workers
1 2 4 8 16 32

性能通常不会无限线性增长。

出现平台期往往意味着：

memory bandwidth saturated

而不是 Shared Memory “失效”。

十四、一个很好用的工程判断公式

如果让我审查一个 Shared Memory 方案，我通常先问四件事：

1. Who owns the memory?

2. Who may write which bytes?

3. Who performs unlink?

4. What happens if any process crashes?

如果这四个问题答不清楚，我通常不会建议直接上线。

真正稳定的结构往往是：

Owner
│
create + metadata
│
┌─────────┼─────────┐
▼ ▼ ▼
Worker 0 Worker 1 Worker 2
│ │ │
slice A slice B slice C
│ │ │
└──────────┼──────────┘
▼
join
│
close/unlink

数据共享，但写入 ownership 不共享。

这恰恰是获得高性能的关键。

十五、总结：Shared Memory 真正的代价是什么？

Shared Memory 最大的优势非常明确：

少复制
少序列化
少 IPC 数据搬运
多个进程直接访问同一份大型数据

但代价同样明确：

生命周期复杂度
ownership 管理
race condition
lock contention
false sharing
cache coherence cost
memory bandwidth contention
crash cleanup
NUMA 问题
调试难度

因此，在 Python 实战中，我并不把 Shared Memory 理解成一个单纯的“性能 API”。

它更像是一份契约：

操作系统允许多个进程共同触碰同一块内存，而作为交换，数据一致性、资源生命周期和访问规则需要由你自己设计。

最理想的 Shared Memory 系统，往往不是拥有最多的锁，而是拥有最清楚的 ownership。

能够只读，就不要并发写。

必须并发写，就做大块分区。

能够由一个 owner 管理生命周期，就不要让每个 worker 都尝试 cleanup。

能够通过架构避免竞争，就不要等性能出现问题之后再到代码里疯狂加锁。

最终你会发现，Shared Memory 最重要的优化可能并不是：

shared_memory.SharedMemory(...)

而是这四个词：

Ownership
Partitioning
Lifecycle
Measurement

把这四件事情设计好，Python 完全可以让多个 worker 高效处理数 GB 甚至更大的 NumPy 数据，同时把复制、同步和故障风险控制在合理范围之内。

而这也正是从“会写 Python 并发代码”，走向“能设计可靠高性能 Python 系统”的分水岭。

留给读者的两个问题

你在实际 Python 项目中，是更倾向于使用 SharedMemory、memmap，还是通过消息队列把数据拆成独立任务？

如果一个 32 核服务器在使用 8 个 worker 后吞吐量就不再提升，你会首先排查 CPU、锁竞争、false sharing，还是内存带宽？

这两个问题的答案，往往比“到底开多少进程最快”更值得讨论。

关键词： Python编程、Python教程、Python实战、Python最佳实践、Python多进程、Shared Memory、NumPy、零拷贝、进程安全、并发编程、性能优化