3. 数值任务的数据共享与通信机制

在 Python 多进程数值计算中，进程之间默认是内存隔离的：每个子进程拥有独立的地址空间，父进程里的 NumPy 数组不会自动被子进程看到。要让多个进程协同处理同一批数值任务，就必须显式选择一种数据共享或通信机制。本节围绕四类常用手段展开：共享内存（Value、Array）、Manager 对象、进程间队列（Queue）与管道（Pipe），重点讨论它们在数值数组传递中的性能差异与常见陷阱。

一、共享内存：Value 与 Array

multiprocessing.Value 和 multiprocessing.Array 基于底层共享内存实现，父子进程直接读写同一块物理内存，不需要序列化拷贝。它们适合“小规模、高频读写”的共享状态，例如全局计数器、累加和、进度标志。

import multiprocessing as mp
import numpy as np

def worker(arr, idx, lock):
    with lock:
        arr[idx] = arr[idx] * 2

if __name__ == "__main__":
    # 'd' 表示 double，10 个元素
    shared = mp.Array('d', 10, lock=True)
    for i in range(10):
        shared[i] = float(i)
    procs = [mp.Process(target=worker, args=(shared, i, shared.get_lock()))
             for i in range(10)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(list(shared))  # 示例输出：[0.0, 2.0, 4.0, ... 18.0]

关键点在于：Array 只支持一维的 C 类型数组，元素类型由类型码（如 'd'、'i'、'f'）决定。它不是 NumPy 数组，但可以通过 np.frombuffer(shared.get_obj()) 零拷贝地映射为 ndarray 视图，从而在数值计算中直接使用。需要注意两点陷阱：其一，多个进程同时写同一元素时必须加锁，否则会出现竞态导致结果不确定；其二，Array 无法直接表达二维或多维结构，需要自行按行主序展平并计算索引，容易出错。

二、Manager 对象：灵活但昂贵

multiprocessing.Manager() 会启动一个独立的服务进程，对外提供 dict、list、Namespace 等代理对象。子进程操作的是代理，实际数据存放在 Manager 进程中，通过 IPC 转发。它的优势是支持任意 Python 对象和嵌套结构，适合共享配置、结果字典、任务登记表等。

def collect(shared_dict, key, value):
    shared_dict[key] = value

if __name__ == "__main__":
    with mp.Manager() as mgr:
        results = mgr.dict()
        procs = [mp.Process(target=collect, args=(results, f"p{i}", i*i))
                 for i in range(4)]
        for p in procs: p.start()
        for p in procs: p.join()
        print(dict(results))  # 示例：{'p0': 0, 'p1': 1, 'p2': 4, 'p3': 9}

性能陷阱非常明显：每次读写代理对象都要经过序列化与进程间往返，单次访问开销远高于本地操作。若在循环中逐元素写入一个 Manager list 来收集百万级数值结果，耗时会比预期高出一到两个数量级。正确做法是让每个子进程先在本地累积成块，最后一次性提交整块结果，而不是高频小粒度访问。

三、队列与管道：面向消息的传递

Queue 是进程安全的 FIFO 队列，支持多生产者多消费者，内部自带锁与缓冲；Pipe 则提供一对连接端点，适合两个进程之间的双向或单向通信，开销略低但拓扑受限。两者传递数据时都会进行 pickle 序列化，因此传递大数组意味着一次完整的拷贝。

def producer(q):
    q.put(np.arange(1000, dtype=np.float64))

def consumer(q):
    data = q.get()
    print(data.sum())  # 示例：499500.0

if __name__ == "__main__":
    q = mp.Queue()
    p1 = mp.Process(target=producer, args=(q,))
    p2 = mp.Process(target=consumer, args=(q,))
    p1.start(); p2.start(); p1.join(); p2.join()

常见陷阱有三个：第一，向 Queue 放入大数组后若未及时取出，数据会滞留在管道缓冲区，可能因缓冲区满而阻塞生产者，甚至引发死锁；第二，Queue 的 join() 依赖 task_done() 配对调用，漏调会导致永久等待；第三，Pipe 两端同时读写同一端点可能造成数据交错，应约定单向使用或加同步。

四、选型与性能判断

把上述机制放到数值数组场景中，可以形成一条清晰的判断链。若数据是“小标量、高频更新”，优先 Value 加锁；若数据是“一维定长数组、需要原地读写”，用 Array 并通过 frombuffer 映射为 ndarray；若需要共享结构化结果或配置，用 Manager，但务必降低访问频次；若任务之间是“生产者—消费者”的消息流，用 Queue；若只有两个进程且通信方向固定，用 Pipe 更轻量。

对于真正的大数组并行计算，更稳妥的模式是“分块 + 共享内存”：父进程用 Array 或 shared_memory 分配一块连续缓冲区，按行切分给各子进程，每个子进程只写自己负责的区间，避免锁竞争；结果汇总时再统一读取。这样既避免了 Queue 的整块序列化拷贝，也避免了 Manager 的高频代理开销。判断是否达成目标，可以观察两点：一是总耗时是否随进程数近似线性下降，二是结果数组与串行版本逐元素比对是否完全一致。若加速比停滞甚至倒退，通常说明序列化或锁竞争已成为瓶颈，应回到机制选型上重新权衡。