无GIL Python中并行随机数生成的性能问题与解决问询

我正在使用Python进行强化学习模拟，需运行大量episode并逐episode更新智能体权重。为加速流程，希望通过并行模拟生成随机数并并行更新权重。

但发现即使使用实验性无GIL的Python 3.13t，随机数生成仍是主要瓶颈：当用N线程运行N个模拟时，耗时是单线程的N倍，仿佛随机数生成未并行化。

复现代码 #

# test.py
import threading
import time
import random

def generate_randoms(n):
    """Generate n random floats between 0 and 1"""
    rng = random.Random()
    for i in range(n):
        rng.random()

def run_tasks_in_parallel(num_tasks, numbers):
    threads = []
    for i in range(num_tasks):
        t = threading.Thread(target=generate_randoms, args=(numbers[i],))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()

if __name__ == "__main__":
    num_tasks = 2
    numbers = [1000000] * 2
    
    start_time = time.time()
    run_tasks_in_parallel(num_tasks, numbers)
    end_time = time.time()
    
    print(f"Time taken: {end_time - start_time} seconds")

基准测试结果（未说明时num_tasks=4） #

• python3.13 test.py：~11s
• python3.13t test.py：~6s
• python3.13t -X gil=0 test.py：~6s
• python3.13t -X gil=1 test.py：~16s
• python3.13 test.py [num_tasks=1]：~1s
• python3.13t test.py [num_tasks=1]：~1s

当num_tasks=2时，程序耗时是num_tasks=1时的2倍。使用numpy.random.default_rng耗时更长。我曾以为是共享随机状态问题，因此为每个线程实例化了独立rng，但问题仍存在。移除随机数生成后并行性正常，已排除Python层面问题。

请问在无GIL Python中如何并行化随机数生成？为何CPython和Numpy的随机数生成无法随线程数扩展？我不需要线程共享随机状态，也不需要线程安全的生成，仅需并行生成随机数以加速训练。

一、为什么随机数生成无法并行扩展？ #

• CPython random模块的底层锁：即使每个线程用独立random.Random实例，其依赖的C扩展实现可能仍存在全局锁或共享资源竞争。3.13t作为无GIL的实验版本，部分C扩展还未完全适配无GIL环境，导致底层操作串行化。
• Numpy的线程安全设计：numpy.random的生成器默认会加锁保证线程安全，即便用独立default_rng实例，某些底层操作仍有隐式锁竞争，无法真正并行。
• 单线程性能与线程开销：随机数生成是CPU密集型操作，若单线程生成速度已接近CPU核心单线程上限，多线程并行时核心资源被争抢，加上调度开销，总耗时会接近单线程的N倍。

二、无GIL Python中并行化随机数生成的可行方案 #

1. 改用多进程（concurrent.futures.ProcessPoolExecutor） #

多进程能彻底隔离每个任务的资源，避免共享状态和锁竞争，每个进程拥有独立解释器和随机数生成器，可真正利用多核CPU。

修改后的示例代码：

import concurrent.futures
import time
import random

def generate_randoms(n):
    rng = random.Random()
    for i in range(n):
        rng.random()

if __name__ == "__main__":
    num_tasks = 2
    numbers = [1000000] * 2
    
    start_time = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        executor.map(generate_randoms, numbers)
    end_time = time.time()
    
    print(f"Time taken: {end_time - start_time} seconds")

2. 使用适配无GIL的随机数库 #

选择无锁设计的第三方库，比如：

• pcg-random：实现PCG随机数算法，无锁并行，适合多线程场景
• randcg：基于C++的无锁随机数生成库

以pcg-random为例，替换random.Random：

from pcgrandom import PCGRandom

def generate_randoms(n):
    rng = PCGRandom()
    for i in range(n):
        rng.random()

3. 预生成批量随机数 #

训练前用单线程或多进程批量生成足够的随机数，存储在数组中，后续模拟直接取用。这种方式能避免训练过程中实时生成的瓶颈，尤其适合强化学习中episode随机需求可预估的场景。

4. 优化Numpy随机数使用 #

若必须用Numpy，确保每个线程/进程的生成器完全独立，避免调用全局随机函数；同时使用批量生成方法（如rng.random(size=n)）减少Python层循环开销，提升单任务效率，间接优化并行效果。

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