Python multiprocessing.shared_memory 如何避免大数组拷贝：生命周期、命名冲突与释放验收

批量处理 NumPy 数组时，最容易被忽略的成本不是计算，而是把几十 MB 数据反复序列化后交给子进程。multiprocessing.shared_memory.SharedMemory 可以把数据放进一块命名共享缓冲区，让其他进程按名称接入；但它不是“创建后自动永远存在”的全局变量，close()、unlink() 和异常清理必须分工明确。

创建者负责建立共享块并在最后一次使用后调用 unlink()，每个附加进程只关闭自己的句柄；验收时同时检查数据、名称和释放结果。

先把共享内存的边界说清楚

SharedMemory 解决的是同一台机器上进程之间的数据共享，数据位于一块可按名称访问的内存区域，不是跨机器的缓存，也不会自动提供数组级锁。它适合大块、结构稳定、读多写少的数据；小对象仍然用进程队列传递往往更简单。

本文示例用 6 个 int64 元素模拟一批待处理数据。为了让子进程重建数组，发送的不是整个数组，而是共享块的 name、元素数量和 dtype 字符串。

创建者与附加者如何走完一条真实调用链

创建进程先分配共享块，再把名称交给子进程。子进程用 SharedMemory(name=name) 附加，基于 shm.buf 创建数组视图，读取完成后关闭自己的句柄。下面的代码故意把清理写进 finally，因为读取失败时也不能把资源留在系统里。

from multiprocessing import Process
from multiprocessing import shared_memory
import numpy as np

def read_shared(name, length, dtype_name):
    shm = shared_memory.SharedMemory(name=name)
    try:
        view = np.ndarray((length,), dtype=np.dtype(dtype_name), buffer=shm.buf)
        print("child data:", view.tolist())
    finally:
        shm.close()

if __name__ == "__main__":
    source = np.array([4, 8, 15, 16, 23, 42], dtype=np.int64)
    shm = shared_memory.SharedMemory(create=True, size=source.nbytes)
    try:
        shared = np.ndarray(source.shape, dtype=source.dtype, buffer=shm.buf)
        shared[:] = source
        worker = Process(target=read_shared,
                         args=(shm.name, source.size, source.dtype.str))
        worker.start()
        worker.join()
        if worker.exitcode != 0:
            raise RuntimeError(f"worker exitcode={worker.exitcode}")
    finally:
        shm.close()
        shm.unlink()

这条链路里，shm.name 是连接凭据，shm.buf 是字节缓冲区，np.ndarray 只是基于缓冲区建立的视图。子进程退出不等于底层共享块已经被删除，所以创建者最后仍要完成 unlink()。

shape 和 dtype 不匹配时，数据看似存在也不可信

共享块只知道字节数，不知道你想把它解释成什么数组。创建端如果传入 6 个 int64，附加端却按 float32 或错误长度构造视图，轻则得到错误结果，重则在构造时直接因为缓冲区大小不足而失败。把元数据和名称放在同一条任务消息中，是比“子进程自己猜”更稳妥的做法。

meta = {
    "name": shm.name,
    "shape": source.shape,
    "dtype": source.dtype.str,
}

# 附加端：
attached = shared_memory.SharedMemory(name=meta["name"])
try:
    restored = np.ndarray(meta["shape"],
                          dtype=np.dtype(meta["dtype"]),
                          buffer=attached.buf)
    assert restored.tolist() == source.tolist()
finally:
    attached.close()

这里的断言不是性能测试，而是接口契约测试：名称能否连接、字节布局能否还原、读取结果是否与写入结果一致，三件事缺一不可。

close 和 unlink 的职责不要混在一起

close() 释放的是当前 SharedMemory 实例持有的文件描述符或句柄；它不会保证底层共享块立即消失。unlink() 才是删除共享块的动作，而且同一个共享块不应由每个进程各调用一次。

POSIX 系统与 Windows 对删除时机的表现不同，因此不要把“调用 unlink 后立刻在所有平台都看不到名称”当作跨平台断言。应用层更可靠的验收是：所有使用者结束后执行一次清理，并记录清理动作是否成功。

命名冲突与异常清理怎么验收

默认让系统生成名称最省心。如果业务必须指定名称，就要把创建和附加区分开：create=True 遇到已存在名称应视为冲突，而不是悄悄覆盖；附加端拿不到名称则应报告“生产者尚未准备好”或“共享块已被清理”，不要无限重试。

name = "orders-batch-20260830"
try:
    shm = shared_memory.SharedMemory(name=name, create=True, size=source.nbytes)
except FileExistsError:
    raise RuntimeError("shared block name collision")

try:
    # 写入、启动消费者、等待结果
    pass
finally:
    shm.close()
    shm.unlink()

测试时至少故意覆盖三个分支：消费者正常退出、消费者非零退出、创建者在消费者启动前抛异常。每个分支都要能在日志中看到 close，并由唯一责任方执行 unlink。

常见问题

SharedMemory 能替代 multiprocessing.Queue 吗？

不能直接替代。共享内存适合大块数据本体，队列仍然适合传递任务、名称和状态消息；两者组合通常比把大数组塞进队列更清楚。

为什么 close 之后共享内存还可能存在？

因为 close 只关闭当前句柄，不负责删除底层共享块。确认最后一个使用者结束后，再由一个明确的责任方调用 unlink。

子进程按名称连接失败怎么办？

先检查生产者是否已经完成 create 并传出了准确的 name，再检查共享块是否已经被提前 unlink；不要先改 size，附加现有块时 size 参数不会修复名称或生命周期问题。

共享 NumPy 数组需要传整个数组吗？

不需要。传递 name、shape 和 dtype 即可在附加端重建视图，但写入和读取必须遵守同一份元数据约定。

落地前的检查清单

• 是否只把大块、结构稳定的数据放进共享内存？
• 是否把 name、shape、dtype 与任务消息绑定传递？
• 每个进程是否在不再访问缓冲区后 close？
• 是否只有一个责任方调用 unlink，并覆盖异常路径？
• 是否用写入可见、附加读取、子进程退出码和清理结果完成验收？