Python RLock 和 Lock 有什么区别?一文讲透重入锁

Python 的 Lock 与 RLock 都能保护多线程共享数据,关键区别在于同一线程能否重复获取同一把锁。本文用可运行示例讲清重入机制、递归计数、死锁边界与实际选型。

7 分钟阅读 Python

Python RLock 和 Lock 的核心区别

Python 里的 LockRLock 都用于保护多线程共享数据,但两者处理“同一个线程重复加锁”的方式不同:

  • Lock 是普通互斥锁。同一线程已经持有锁后再次获取,会像其他线程一样等待,容易把自己锁死。
  • RLock 是可重入锁。同一线程可以重复获取,内部会记录持有者和获取次数;只有等量释放后,其他线程才能拿到锁。
对比项 threading.Lock threading.RLock
同一线程能否重复获取 不能,会阻塞或获取失败 可以,会增加递归层级
是否记录持有线程 不提供所有权约束 记录持有线程
释放要求 锁定后可由任意线程调用 release() 必须由持有锁的线程释放
多次获取后的释放 不适用 获取几次就要释放几次
典型场景 简单、单层临界区 嵌套调用、递归调用、可重入 API
选型建议 默认优先 确实存在同线程嵌套加锁时使用

一句话判断:只加一层锁,用 Lock;同一条调用链可能反复进入受保护代码,用 RLock

Lock 为什么会把当前线程锁住

Lock 只有“未锁定”和“已锁定”两种状态,不关心当前请求锁的线程是不是原持有者。下面的代码用超时参数演示同一线程重复获取,避免示例永久卡住:

import threading

lock = threading.Lock()

with lock:
    acquired_again = lock.acquire(timeout=0.2)
    print(acquired_again)  # False

进入 with lock 后,当前线程已经持有锁。再次调用 acquire() 时,普通 Lock 不会因为“还是这个线程”就放行。示例设置了 timeout=0.2,因此等待约 0.2 秒后返回 False;如果不设超时,线程会一直等待。

这种问题经常出现在方法嵌套中:

import threading


class Counter:
    def __init__(self) -> None:
        self._value = 0
        self._lock = threading.Lock()

    def increment(self) -> None:
        with self._lock:
            self._value += 1

    def increment_twice(self) -> None:
        with self._lock:
            self.increment()
            self.increment()

increment_twice() 先拿到锁,再调用同样会拿锁的 increment()。如果执行 increment_twice(),当前线程会在第一次调用 increment() 时阻塞。这不是线程之间互相等待,而是线程在等待自己已经持有的锁。

RLock 为什么可以重复加锁

RLock 比普通锁多维护两类信息:

  1. 当前由哪个线程持有;
  2. 当前线程重复获取了多少次,也就是递归层级。

同一线程再次调用 acquire() 时,RLock 不会阻塞,只会把递归层级加一。每次 release() 会让层级减一,只有降到零时才真正解锁。

把上面的 Counter 改成 RLock,嵌套调用就能正常完成:

import threading


class Counter:
    def __init__(self) -> None:
        self._value = 0
        self._lock = threading.RLock()

    def increment(self) -> None:
        with self._lock:
            self._value += 1

    def increment_twice(self) -> None:
        with self._lock:
            self.increment()
            self.increment()

    @property
    def value(self) -> int:
        with self._lock:
            return self._value


counter = Counter()
counter.increment_twice()
print(counter.value)  # 2

这里一共进入了三次受保护代码:increment_twice() 一次,两个 increment() 各一次。每个 with 退出时都会自动释放一层,最外层退出后锁才真正可供其他线程获取。

什么时候应该用 RLock

1. 公共方法会调用另一个需要加锁的公共方法

一个类的多个公共方法都要保证线程安全,其中某个组合方法又会调用其他公共方法,这正是 RLock 最常见的使用场景。每个方法都能独立保证安全,组合调用也不会自锁。

2. 递归函数需要保护同一份共享状态

递归遍历树、图或嵌套结构时,每一层可能进入同一个受保护函数。如果所有递归层共享一把锁,普通 Lock 会在第二层阻塞,RLock 可以跟踪递归深度。

3. 回调或框架钩子可能重新进入当前对象

受保护方法调用回调后,回调又回到同一对象的另一个加锁方法,这种路径不一定能从单个函数看出来。只要确认重入是合理业务行为,就可以使用 RLock

不过,RLock 不应该成为“发现死锁就替换”的万能补丁。如果两个线程以不同顺序获取两把锁,例如线程 A 先拿锁 1 再等锁 2,线程 B 先拿锁 2 再等锁 1,换成 RLock 仍然会死锁。它只解决同一线程重复获取同一把锁的问题。

什么时候普通 Lock 更合适

如果临界区结构简单,没有嵌套加锁,优先使用 Lock

import threading

total = 0
lock = threading.Lock()


def add_one() -> None:
    global total
    with lock:
        total += 1

选择 Lock 的理由不是追求微小的性能差异,而是语义更直接:这段代码不允许重入。如果未来意外出现嵌套获取,问题会尽早暴露,而不是被递归计数掩盖。

可以先问自己两个问题:

  1. 当前线程持有锁时,会不会直接或间接再次进入同一把锁保护的代码?
  2. 这种重入是设计需要,还是职责耦合过深造成的?

第一题为“否”,用 Lock;第一题为“是”且第二题确认是合理设计,再用 RLock

Lock 和 RLock 都要避开的坑

手动加锁后忘记释放

优先使用上下文管理器:

with lock:
    update_shared_state()

它能在代码块抛出异常时自动释放锁,比手写 acquire()release() 更安全。对 RLock 来说,它还能保证每次进入都对应一次退出,减少递归层级无法归零的问题。

持锁执行网络或磁盘 I/O

锁的范围越大,其他线程等待越久。网络请求、文件读写或数据库访问可能耗时且不可控,通常应先在锁外完成慢操作,再用尽可能短的临界区更新共享状态。

以为有 GIL 就不需要锁

GIL 不等于业务数据自动线程安全。复合操作可能包含多个步骤,线程可能在步骤之间切换;阻塞 I/O 期间,CPython 也会释放 GIL。此外,Python 还提供可自由线程化构建。只要多个线程会读写同一份可变状态,就应该按共享状态的不变量设计同步方案,而不是依赖解释器实现细节。

获取 RLock 多次,却少释放一次

如果手动获取 RLock 三次,就必须由持有它的线程释放三次。少释放一次,递归层级就不会回到零,其他线程仍然无法获取。无法使用 with 时,至少用 try...finally 保证成对释放。

常见问题

RLock 比 Lock 慢吗?

RLock 需要跟踪持有线程和递归层级,机制比 Lock 更复杂。但实际选型应先看语义是否需要重入,不要在没有基准测试的情况下为了猜测的性能差异牺牲正确性。临界区大小、锁竞争程度和持锁时间通常更值得关注。

RLock 能防止所有死锁吗?

不能。它只能避免同一线程再次获取同一把锁时的自锁。锁顺序不一致、持锁等待外部资源、线程互相等待等问题仍然可能造成死锁。

Lock 可以由别的线程释放吗?

可以。Python 官方文档明确说明,threading.Lock.release() 可以由任意线程调用,只要锁当前处于锁定状态。RLock 不同,它只能由持有锁的线程释放。虽然 Lock 支持跨线程释放,普通业务代码仍应让获取与释放处于清晰、可审计的控制流中。

Condition 默认使用哪种锁?

创建 threading.Condition() 时如果不传入锁,Python 会创建一把 RLock 作为底层锁。你也可以显式传入 LockRLock,但要确保它符合调用链是否重入的要求。

选型结论

  • 临界区简单、不会嵌套获取同一把锁:使用 Lock
  • 同一线程会通过嵌套方法、递归或回调再次获取同一把锁:使用 RLock
  • 不确定时先梳理调用链和锁的边界,不要直接用 RLock 掩盖结构问题。
  • 两种锁都优先配合 with 使用,并缩短持锁时间。

进一步核对 API 行为,可以查看 Python 官方 threading 文档;如果要理解 GIL 与线程状态的关系,可参考 Python/C API 线程状态与 GIL 文档

Practice

读完这一节,去靶场里验证一下。

去挑战广场练习