Python multiprocessing共享内存不生效，testfunc2无法修改共享数组是什么原因？

你的代码存在四个核心问题：

• mp.Array创建的是固定长度的共享ctypes数组，本身不支持动态扩容。你在testfunc1中执行arr = np.concatenate((arr, np.array([0])))时，numpy会自动将传入的mp.Array包装对象转为普通numpy数组，拼接后生成的新数组也是普通numpy数组，赋值给局部变量arr后，你后续操作的完全是当前进程内的局部变量，和原共享内存没有任何关联，更不会影响其他进程。
• 多进程的地址空间互相独立，你在testfunc1中修改局部变量arr的指向，仅对当前进程生效，testfunc2中的arr仍然指向原始的共享数组对象，两个进程的arr变量在赋值后已经完全无关。
• testfunc2中直接打印arr对象，输出的是SynchronizedArray包装类的元信息，不是数组的实际内容。你看不到testfunc2的修改效果，本质是因为testfunc1已经不再读取共享内存的数据，只操作自己的局部数组。
• 你没有使用mp.Array自带的锁保证并发操作的原子性，即使修复了上述问题，并发读写也可能出现数据异常。

首先明确：如果你需要动态扩容的共享数组，mp.Array不适用，可改用multiprocessing.Manager提供的list结构；如果不需要动态扩容，可直接基于固定长度的mp.Array修改，以下是两种场景的实现示例：

场景1：不需要动态扩容，仅验证共享修改效果 #

import numpy as np
import multiprocessing as mp
import time

# 创建固定长度的共享数组
arr = mp.Array('f', np.array([1,2,3,4]))

def testfunc1(arr):
    while True:
        # 加锁后读取共享数组内容，转为numpy数组使用
        with arr.get_lock():
            np_arr = np.array(arr[:])
        print("testfunc1 读取结果：", np_arr)
        time.sleep(2)

def testfunc2(arr):
    while True:
        # 加锁保证修改操作的原子性
        with arr.get_lock():
            arr[0] *= 2
            np_arr = np.array(arr[:])
        print("testfunc2 修改结果：", np_arr)
        time.sleep(2)

if __name__ == "__main__":
    proc1 = mp.Process(target=testfunc1, args=(arr,))
    proc2 = mp.Process(target=testfunc2, args=(arr,))
    proc1.start()
    proc2.start()
    proc1.join()
    proc2.join()

场景2：需要动态新增元素，改用Manager管理的共享列表 #

import numpy as np
import multiprocessing as mp
import time

def testfunc1(shared_list, lock):
    while True:
        with lock:
            shared_list.append(0)
            np_arr = np.array(shared_list)
        print("testfunc1 新增元素结果：", np_arr)
        time.sleep(2)

def testfunc2(shared_list, lock):
    while True:
        with lock:
            if len(shared_list) > 0:
                shared_list[0] *= 2
            np_arr = np.array(shared_list)
        print("testfunc2 修改结果：", np_arr)
        time.sleep(2)

if __name__ == "__main__":
    # 创建支持动态扩容的共享列表
    manager = mp.Manager()
    shared_list = manager.list([1,2,3,4])
    # 创建全局锁保证并发安全
    lock = manager.Lock()
    proc1 = mp.Process(target=testfunc1, args=(shared_list, lock))
    proc2 = mp.Process(target=testfunc2, args=(shared_list, lock))
    proc1.start()
    proc2.start()
    proc1.join()
    proc2.join()

• 多进程代码必须加if __name__ == "__main__"的入口判断，Windows系统下不加会直接报错，类Unix系统下也建议加上避免异常。
• 所有对共享资源的读写操作都要加锁，避免竞态条件导致的数据错乱。
• 不要给接收共享对象的局部变量重新赋值，否则会丢失对共享资源的引用。

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