如何在父进程与子进程间共享NumPy随机状态？

这个问题我之前也踩过坑——multiprocessing的Process在启动时（Unix下是fork，Windows下是spawn）会复制父进程的整个内存空间，包括NumPy的随机状态数组。这就导致每个子进程一开始的随机状态和父进程完全相同，生成的随机序列自然重复。要实现真正的状态共享，核心是让所有进程操作同一个可共享的状态存储，同时处理好进程间的竞争冲突。

下面给你两种实用的解决方案：

方案1：使用共享内存+锁（性能优先） #

这种方案直接用multiprocessing.Array创建共享内存存储随机状态的核心数组，配合锁来避免多进程同时修改状态导致的竞争问题，性能比下一种方案更好，适合频繁生成随机数的场景。

import numpy as np
import multiprocessing as mp

# 全局变量用于子进程存储共享状态和锁
np_rng_state = None
state_lock = None
rng_type = None
has_gauss = None
cached_gauss = None

def init_shared_state(shared_arr, lock):
    """初始化子进程的共享状态引用"""
    global np_rng_state, state_lock, rng_type, has_gauss, cached_gauss
    # 将共享内存数组转为NumPy数组（只读视图，修改会同步到共享内存）
    np_rng_state = np.frombuffer(shared_arr.get_obj(), dtype=np.uint32)
    state_lock = lock
    # 从父进程初始化状态参数
    parent_state = np.random.get_state()
    rng_type, _, has_gauss, cached_gauss = parent_state
    np_rng_state[:len(parent_state[1])] = parent_state[1]

def worker(shared_arr, lock):
    """子进程函数：生成随机数并更新共享状态"""
    init_shared_state(shared_arr, lock)
    
    def get_shared_random():
        """线程安全的共享随机数生成函数"""
        global np_rng_state, rng_type, has_gauss, cached_gauss, state_lock
        with state_lock:
            # 恢复当前共享状态
            current_state = (rng_type, np_rng_state.copy(), has_gauss, cached_gauss)
            np.random.set_state(current_state)
            # 生成随机数
            val = np.random.rand()
            # 更新共享状态到内存
            new_state = np.random.get_state()
            np_rng_state[:len(new_state[1])] = new_state[1]
            has_gauss, cached_gauss = new_state[2], new_state[3]
        return val
    
    # 测试生成3个随机数
    for _ in range(3):
        print(f"Worker {mp.current_process().name} generated: {get_shared_random():.6f}")

if __name__ == "__main__":
    # 设置父进程初始随机种子
    np.random.seed(42)
    # 获取父进程的随机状态，确定状态数组长度
    parent_state = np.random.get_state()
    state_array_len = len(parent_state[1])
    
    # 创建共享内存数组（类型为uint32，和NumPy随机状态数组类型一致）
    shared_state_arr = mp.Array('I', state_array_len)
    # 创建进程锁，保护状态的读写
    state_lock = mp.Lock()
    
    # 启动2个子进程
    processes = [mp.Process(target=worker, args=(shared_state_arr, state_lock)) for _ in range(2)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    # 父进程也可以使用共享状态生成随机数
    with state_lock:
        np.random.set_state((parent_state[0], np.frombuffer(shared_state_arr.get_obj(), dtype=np.uint32), parent_state[2], parent_state[3]))
        print(f"Parent process generated: {np.random.rand():.6f}")

关键点说明： #

• 用mp.Array创建的共享内存可以被所有进程直接访问，修改会同步到内存中
• 必须用锁（mp.Lock）包裹状态的读写操作，避免多进程同时修改导致状态错乱
• NumPy的随机状态是一个元组：(rng_type, state_array, has_gauss, cached_gauss)，我们只需要共享state_array，其他参数可以从父进程初始化后保持一致

方案2：使用multiprocessing.Manager（简洁优先） #

如果你的场景对性能要求不高，multiprocessing.Manager可以帮你简化共享对象的管理，不需要手动处理共享内存的转换，代码更简洁：

import numpy as np
import multiprocessing as mp

def worker(shared_state, lock):
    """子进程函数：使用Manager共享的状态生成随机数"""
    with lock:
        # 恢复共享状态
        np.random.set_state(shared_state['current_state'])
        # 生成随机数
        val = np.random.rand()
        # 更新共享状态
        shared_state['current_state'] = np.random.get_state()
    print(f"Worker {mp.current_process().name} generated: {val:.6f}")

if __name__ == "__main__":
    # 设置初始种子
    np.random.seed(42)
    
    # 使用Manager创建可共享的字典存储随机状态
    manager = mp.Manager()
    shared_state = manager.dict()
    shared_state['current_state'] = np.random.get_state()
    # 创建锁保护状态读写
    state_lock = mp.Lock()
    
    # 启动子进程
    processes = [mp.Process(target=worker, args=(shared_state, state_lock)) for _ in range(2)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    # 父进程使用共享状态生成随机数
    with state_lock:
        np.random.set_state(shared_state['current_state'])
        print(f"Parent process generated: {np.random.rand():.6f}")

关键点说明： #

• Manager会在后台启动一个服务进程，负责管理共享对象的序列化和传递，所以代码更简洁
• 同样需要锁来保证状态操作的原子性，避免竞争问题
• 性能略低于直接使用共享内存，适合随机数生成频率不高的场景

注意事项 #

1. Windows系统特殊处理：Windows下multiprocessing使用spawn而不是fork，所以所有子进程需要的全局变量和初始化代码必须放在if __name__ == "__main__"块内部，避免重复初始化
2. 锁的必要性：无论哪种方案，都必须使用锁来保护随机状态的读写，否则多进程同时修改会导致状态损坏，生成的随机数序列不可预测
3. 需求匹配：如果你只是想避免子进程生成重复的随机数，不需要共享状态，那么可以让父进程给每个子进程分配独立的种子（比如用np.random.randint生成不同的种子传递给子进程），这种方案更简单

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