np.random.choice结合multiprocessing在for循环中报IndexError求解

你遇到的报错和并行异常来自3个核心逻辑错误，和np.random.choice本身无关：

1. 全局变量传递失效：multiprocessing启动子进程时，子进程会加载模块最顶层的变量定义，你在if __name__ == '__main__'块里给X赋值的操作只在主进程生效，子进程里的X始终是最开头定义的空列表[]，执行X[a]时空列表取索引自然触发IndexError: list index out of range。串行for循环跑在主进程内，能读到main块里赋值的X，所以不会报错。
2. 随机数初始化无效：你写的np.random.RandomState(k)只是创建了一个随机数实例，既没有绑定到numpy全局随机流，也没有用这个实例调用choice方法，等于完全没生效，子进程的随机状态不受你控制。
3. 进程池反复创建浪费资源：你在for循环里每次迭代都新建、销毁进程池，进程启动销毁的额外开销会抵消并行带来的速度提升。

两种常用实现方式，第一种是直接传参，写法最简单：

from multiprocessing import Pool
import numpy as np

K = 10

def function(args):
    k, X = args
    # 正确初始化当前进程的随机数生成器，用k做种子保证可复现
    rng = np.random.default_rng(seed=k)
    aux = [i for i in np.arange(K) if i != k]
    a,b,c = rng.choice(aux, 3, replace=False)
    x = X[a] + 0.7*(X[b]-X[c])
    return x

if __name__ == '__main__':
    X = np.arange(K)
    # 进程池只创建一次，不要放在循环里反复初始化
    pool = Pool(processes=K)
    for n in range(K):
        # 把需要的参数打包传给worker函数
        task_args = [(k, X) for k in np.arange(K)]
        w = pool.map(function, task_args)
        print(w)
    pool.close()
    pool.join()

如果不想每次传X，也可以用进程池初始化器给子进程设置全局变量，适合大数组不想反复拷贝参数的场景：

from multiprocessing import Pool
import numpy as np

K = 10
X = []

def init_worker(shared_X):
    # 子进程初始化时把全局X替换成传入的值
    global X
    X = shared_X

def function(k):
    rng = np.random.default_rng(seed=k)
    aux = [i for i in np.arange(K) if i != k]
    a,b,c = rng.choice(aux, 3, replace=False)
    x = X[a] + 0.7*(X[b]-X[c])
    return x

if __name__ == '__main__':
    X = np.arange(K)
    # 初始化进程池时传入全局变量
    pool = Pool(processes=K, initializer=init_worker, initargs=(X,))
    for n in range(K):
        w = pool.map(function, np.arange(K))
        print(w)
    pool.close()
    pool.join()

• 不要依赖跨进程的全局变量隐式传值，spawn模式下子进程完全重新加载模块，顶层定义之外的变量不会自动同步，fork模式虽然会拷贝父进程内存，但主进程后续修改变量也不会同步到子进程，非常容易出隐蔽bug。
• numpy新版本推荐用np.random.default_rng()创建独立生成器实例，不要依赖全局随机流，多进程下每个进程用独立生成器、独立种子可以避免随机数重复的问题。
• 进程池是重量级资源，初始化一次即可，循环内反复创建销毁会带来不必要的性能损耗。

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