直接用 threading.Lock() 易出错,因异常、提前 return 或逻辑分支可能导致 release() 被跳过而死锁;多方法共用锁时还易遗漏读操作的一致性保障。

为什么直接用 threading.Lock() 容易出错
手动在方法开头 lock.acquire()、结尾 lock.release() 看似简单,但一旦方法里抛异常、提前 return 或逻辑分支多,release() 就可能被跳过,导致死锁。这不是小概率事件——尤其在有数据库操作、网络调用或条件判断的业务方法里,漏掉 release() 是常见故障源。
更麻烦的是,多个方法共用一把锁时,容易误判加锁范围:比如只给写操作加锁,却忘了读操作也需要一致性保障(如检查后再设置的“check-then-set”逻辑)。
用 contextlib.contextmanager 封装锁最稳妥
把加锁/释放逻辑收进上下文管理器,靠 Python 的 with 语句保证无论是否异常都释放锁。不用改类结构,也不依赖继承,复用性高。
示例:一个线程安全的计数器类
import threading
from contextlib import contextmanager
<p>class ThreadSafeCounter:
def <strong>init</strong>(self):
self._value = 0
self._lock = threading.Lock()</p><pre class="brush:php;toolbar:false;">@contextmanager
def _locked(self):
self._lock.acquire()
try:
yield
finally:
self._lock.release()
def increment(self):
with self._locked():
self._value += 1
def get(self):
with self._locked():
return self._value
-
_locked()是私有上下文管理器,隔离锁细节,避免子类或外部误用acquire()/release() - 所有访问
self._value的方法都必须走with self._locked():,否则不安全 - 不要在
_locked()外部暴露self._lock,防止被绕过
何时该用 threading.RLock() 而不是 Lock
当类内部方法存在调用嵌套(比如 reset() 调用 set(value),而两者都需要加锁),普通 Lock 会直接阻塞自己——因为同一线程重复 acquire() 会卡住。
这时换成 RLock(可重入锁):
def __init__(self):
self._lock = threading.RLock() # 注意这里
-
RLock允许同一线程多次acquire(),只要对应次数的release() - 性能略低于
Lock,但对嵌套调用是刚需;若确定无嵌套,优先用Lock - 仍需配合
contextmanager使用,RLock不解决异常漏释放问题
别忽略锁粒度与性能陷阱
锁太粗(比如整个对象一把锁)会严重限制并发,锁太细(每个字段单独锁)又增加维护复杂度和死锁风险。实际中建议:
- 先明确「哪些状态变更必须原子」:比如银行账户的转账,
from_balance扣减和to_balance增加必须一起成功或失败,不能拆成两个锁 - 避免在锁内做耗时操作:如调用
requests.get()、读写文件、序列化大对象——这会让其他线程长时间等待 - 如果只是读多写少,考虑
threading.local()或concurrent.futures.ThreadPoolExecutor分离数据副本,而非硬扛锁
自动加锁不是银弹,关键在厘清共享状态边界。锁本身不难写,难的是判断哪里真需要锁、哪里其实可以无锁化。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











