导读:本期聚焦于毕达哥创作的《Python 如何安全地在多线程里使用 random(不加锁)》,敬请观看详情。CPython 中 random 模块的全局函数共享同一个 Random 实例,底层 C 方法虽然受 GIL 保护使单次调用原子,但 shuffle、sample 等多步骤操作会因线程切换破坏状态一致性。本文从 Mersenne Twister 状态更新机制出发,分析官方文档不建议在并发环境共享全局 random 的原因;然后给出真正无需显式加锁的两种方案:借助 threading.local 为每个线程维护独立 Random 实例,或使用基于 os.urandom 的 SystemRandom。文章包含可运行示例、种子初始化建议,以及不同方案在性能和安全等级上的对比。读完能明确判断在 Web 请求、任务队列、模拟计算等场景下应选择哪种随机数生成方式,避免出现重复序列或统计偏差。

CPython 解释器在执行 random.random() 时,底层 _random.Random 的 C 函数并不会主动释放 GIL,因此从字节码层面看,单个 random() 调用像是一个原子操作。但很多项目因此把模块级 random 函数直接扔进多线程使用,结果出现随机序列重复、统计分布异常等问题。要解释这个现象,必须回到 random 模块的内部状态管理。

Python 如何安全地在多线程里使用 random(不加锁)

全局 random 函数为什么不是线程安全的

random 模块在导入时会创建一个隐藏的 Random 实例,所有模块级便捷函数,例如 random.random()random.randint()random.shuffle(),都绑定到这个实例上。也就是说,无论代码里创建多少个线程,只要调用模块级函数,它们实际上都在访问同一个 Mersenne Twister 状态数组。

在 CPython 中,C 层实现的 random() 方法因为持有 GIL,所以单个方法调用期间不会被其他线程切换;但 shuffle()sample()choices() 等 Python 层实现往往需要连续多次调用底层随机方法。线程可能在两次调用之间切换,另一个线程也来读取或推进同一个状态,造成交叉推进。这不会直接导致 Python 崩溃,但会破坏随机序列的独立性和均匀性,甚至出现两个线程返回相同随机数的情况。

官方文档明确指出,Random 类的实例不是线程安全的,多线程程序应当为每个线程创建独立的实例,避免共享状态。这里需要注意,官方描述并不包含显式锁,因为加锁虽然能保证调用串行化,但会引入锁竞争,并降低吞吐。更好的做法是直接消除共享,也就是每个线程拥有自己的生成器。

不加锁方案一:用 threading.local 给每个线程独立 Random

消除共享的最常见方式是使用 threading.local。它不是锁,而是线程本地存储,每个线程访问同一个属性名时得到各自不同的对象。把 random.Random 实例放在线程本地存储中,即可保证线程之间不共享任何状态,自然不需要显式加锁。

下面是一个可运行的封装示例:

import random
import threading

_local = threading.local()

def get_rng():
    rng = getattr(_local, 'rng', None)
    if rng is None:
        rng = random.Random()
        _local.rng = rng
    return rng

def worker(name):
    rng = get_rng()
    values = [rng.randint(1, 1000) for _ in range(5)]
    print(name, values)

threads = []
for i in range(4):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

这里每个线程第一次调用 get_rng() 时都会创建自己的 Random 实例,后续调用直接复用,实现了与全局 random 相同的 API,却不产生共享状态。因为不同线程的随机数生成器完全独立,即使同时运行也不会有竞态。这样做的额外开销只是每个线程多保存一个对象,比加锁更轻量。

需要特别注意的是种子初始化。如果多个线程都使用默认种子,它们会生成完全相同的随机序列。生产环境最好使用 os.urandom 或当前时间与线程标识组合来设置种子。例如可以这样初始化:

import os
import random
import threading

_local = threading.local()

def get_rng():
    rng = getattr(_local, 'rng', None)
    if rng is None:
        seed = int.from_bytes(os.urandom(8), 'big')
        rng = random.Random(seed)
        _local.rng = rng
    return rng

使用 os.urandom 能获得不可预测的种子,避免多线程生成相同序列。不过要清楚,random.Random 本身不是密码学安全的随机源,即使种子随机,生成序列也不应用于安全令牌或加密密钥。

不加锁方案二:使用 SystemRandom 获得无状态安全

另一种思路是改用 random.SystemRandom。它的底层实现基于 os.urandom(),由操作系统提供的 CSPRNG 生成随机字节,内部不维护类似 Mersenne Twister 那样的可变状态。正因为它没有需要推进的连续状态,多个线程同时调用时不会发生交叉污染,因此可以共享同一个 SystemRandom 实例而无需显式加锁。

使用方式非常简单:

import random

crypto_rng = random.SystemRandom()

def worker():
    for _ in range(10):
        value = crypto_rng.randint(1, 100)
        # 并发安全,无需加锁

threads = []
for _ in range(8):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

不过 SystemRandom 的随机质量虽然高,性能却明显低于 Mersenne Twister。操作系统熵源调用频率过高时可能成为瓶颈,不适合每秒百万次随机数的模拟计算。它更适合生成会话 ID、临时密码、验证码等安全敏感场景。如果只是蒙特卡洛模拟、游戏随机掉落、抽样等对统计质量要求高但对密码学安全没有要求的场景,线程本地 Random 实例是更合适的选择。

有一点需要强调,SystemRandom 的线程安全性依赖于 os.urandom 的系统调用实现,在主流操作系统上都是线程安全的。但不能把这种安全性类推到所有 random.Random 子类。任何自己实现了内部状态、并且多个方法依赖多个步骤更新状态的生成器,都要按共享状态处理。

Python random多线程安全ThreadLocal修改时间:2026-08-20 16:16:08

免责声明:​ 已尽一切努力确保本网站所含信息的准确性。网站内容多为原创整理与精心编撰,观点力求客观中立。本站旨在免费分享,内容仅供个人学习、研究或参考使用。若引用了第三方作品,版权归原作者所有。如内容涉及您的权益,请联系我们处理。
内容垂直聚焦
专注技术核心技术栏目,确保每篇文章深度聚焦于实用技能。从代码技巧到架构设计,为用户提供无干扰的纯技术知识沉淀,精准满足专业提升需求。
知识结构清晰
覆盖从开发到部署的全链路。AI、前端、编程、数据库、服务器、建站、系统层层递进,构建清晰学习路径,帮助用户系统化掌握开发与运维所需的核心技术。
深度技术解析
拒绝泛泛而谈,深入技术细节与实践难点。无论是数据库优化还是服务器配置,均结合真实场景与代码示例进行剖析,致力于提供可直接应用于工作的解决方案。
专业领域覆盖
精准对应开发生命周期。从前端界面到后端编程,从数据库操作到服务器运维,形成完整闭环,一站式满足全栈工程师和运维人员的技术需求。
即学即用高效
内容强调实操性,步骤清晰、代码完整。用户可根据教程直接复现和应用于自身项目,显著缩短从学习到实践的距离,快速解决开发中的具体问题。
持续更新保障
专注既定技术方向进行长期、稳定的内容输出。确保各栏目技术文章持续更新迭代,紧跟主流技术发展趋势,为用户提供经久不衰的学习价值。