使用multiprocessing共享内存写操作的冲突问题求解

你的问题本质是竞态条件（Race Condition）：多个进程同时读取共享内存中的arr[0,0,0]值，各自完成加1操作后再写回，导致后写的进程直接覆盖先写的结果，最终出现重复值。比如进程A和B同时读到0，加1后都得到1，写回后就会出现两个1。

下面是两种实用的解决办法：

方法一：用进程锁（Lock）同步访问 #

通过multiprocessing.Manager创建跨进程共享的锁，保证同一时间只有一个进程能修改共享内存的目标元素。修改后的代码如下：

import numpy as np
from multiprocessing import shared_memory, Pool, Manager
from itertools import repeat
import time


def test_shm(N=500, n_proc=8, name='example'):
    # 创建共享内存数组
    a = np.random.rand(N, N, N).astype(np.float64)
    a[0, 0, 0] = 0.0

    shm = shared_memory.SharedMemory(name=name, create=True, size=a.nbytes)
    b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
    b[:] = a[:]
    shm.close()

    # 创建跨进程共享的锁
    with Manager() as manager:
        lock = manager.Lock()
        with Pool(n_proc) as p:
            res = p.starmap(
                work, zip(range(n_proc), repeat(name), repeat(a.dtype), repeat(N), repeat(lock)))

        for r in res:
            print(f'{r[0]}\t{r[1]}')

        res = np.array([r[0] for r in res])
        print('not ' * int(~np.all(np.sort(res) == 1 + np.arange(n_proc))) + 'all good')

    shm.unlink()


def work(i, name, dtype, N=500, lock=None):
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray((N, N, N), dtype=dtype, buffer=shm.buf)
    time.sleep(2)
    
    # 加锁保护共享内存的修改操作
    with lock:
        val = arr[0, 0, 0:2].copy()
        val[0] += 1.0
        arr[0, 0, 0] = val[0]
    
    shm.close()
    return val


if __name__ == '__main__':
    test_shm()

with lock:上下文会自动完成锁的获取和释放，确保读取-修改-写入的操作链完全串行化，彻底避免竞态条件。

方法二：用multiprocessing.Value管理原子变量 #

如果只是需要对单个数值做原子更新，可以用Value替代直接操作numpy数组元素——Value内部自带锁，支持原子操作。修改后的代码示例：

import numpy as np
from multiprocessing import shared_memory, Pool, Manager, Value
from itertools import repeat
import time


def test_shm(N=500, n_proc=8, name='example'):
    # 创建共享内存数组
    a = np.random.rand(N, N, N).astype(np.float64)
    a[0, 0, 0] = 0.0

    shm = shared_memory.SharedMemory(name=name, create=True, size=a.nbytes)
    b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
    b[:] = a[:]
    shm.close()

    # 创建共享的浮点型Value，初始值0.0
    with Manager() as manager:
        shared_val = manager.Value('d', 0.0)
        with Pool(n_proc) as p:
            res = p.starmap(
                work, zip(range(n_proc), repeat(name), repeat(a.dtype), repeat(N), repeat(shared_val)))

        for r in res:
            print(f'{r[0]}\t{r[1]}')

        res = np.array([r[0] for r in res])
        print('not ' * int(~np.all(np.sort(res) == 1 + np.arange(n_proc))) + 'all good')

    shm.unlink()


def work(i, name, dtype, N=500, shared_val=None):
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray((N, N, N), dtype=dtype, buffer=shm.buf)
    time.sleep(2)
    
    # 原子更新Value，再同步到numpy数组
    with shared_val.get_lock():
        shared_val.value += 1.0
        val_updated = shared_val.value
        arr[0, 0, 0] = val_updated
    
    val = arr[0, 0, 0:2].copy()
    val[0] = val_updated
    shm.close()
    return val


if __name__ == '__main__':
    test_shm()

Value('d', 0.0)创建了一个双精度浮点型的共享变量，get_lock()获取内部锁，保证+=1.0是原子操作，不会被其他进程打断，之后再把更新后的值同步到numpy数组即可。

注意事项 #

• 锁会带来一定性能开销，但这是保证多进程共享资源数据一致性的必要代价，实际生产场景必须通过同步机制解决这类问题。
• 如果你的场景以只读操作为主，仅少数写操作，尽量缩小锁的作用范围，只在写操作的代码块上加锁，减少对整体性能的影响。

内容的提问来源于stack exchange，提问作者John Smith