使用Multiprocessing结合Numpy生成随机数的代码正确性验证

原代码的正确性说明 #

你当前的代码可以正常工作，核心逻辑没问题：多进程启动时会复制主进程的随机状态，因此必须在子进程中重新初始化随机种子，否则所有进程会生成完全相同的随机数。但np.random.seed()属于Numpy旧版的全局随机状态API，确实已被官方弃用，虽然目前还能运行，但长期维护建议切换到官方推荐的Generator API。

使用Generator API的替代方案 #

Generator是Numpy 1.17+推荐的随机数生成方式，它基于独立的随机数生成器实例，避免全局状态的问题，在多进程场景下更安全可控。以下是两种可行的实现方式：

方式1：子进程独立初始化Generator #

每个子进程创建自己的Generator实例，用系统熵初始化种子（和原代码np.random.seed()的效果类似，但更规范）：

import numpy as np
from multiprocessing import Pool

def generate_random(iproc):
    # 创建独立的Generator实例，自动用系统熵初始化种子
    rng = np.random.default_rng()
    nums = rng.uniform(size=8)
    print(f'{iproc=},{nums=}')

if __name__ == '__main__':
    nproc = 8
    arglist = [(iproc,) for iproc in range(nproc)]
    with Pool(nproc) as p:
        p.starmap(generate_random, arglist)

方式2：主进程预生成种子分发给子进程（更可控） #

如果需要可复现的随机结果，可以在主进程生成每个子进程的种子，再传递给子进程初始化Generator：

import numpy as np
from multiprocessing import Pool

def generate_random(iproc, seed):
    # 用主进程传递的种子初始化Generator
    rng = np.random.default_rng(seed)
    nums = rng.uniform(size=8)
    print(f'{iproc=},{nums=}')

if __name__ == '__main__':
    nproc = 8
    # 主进程生成一组独立的种子
    main_rng = np.random.default_rng()
    seeds = main_rng.integers(0, 2**32, size=nproc)
    arglist = [(iproc, seeds[iproc]) for iproc in range(nproc)]
    with Pool(nproc) as p:
        p.starmap(generate_random, arglist)

总结 #

• 原代码逻辑正确，但依赖已弃用的API，建议迁移到Generator
• default_rng()是官方推荐的生成Generator实例的方法，无参数时自动用系统熵初始化，避免全局状态冲突
• 如果需要可复现的结果，主进程预生成种子的方式更可靠

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