多线程编程中，共享数据是万恶之源。Python 的 threading 模块提供了一套锁工具，核心目的只有一个：让并发代码串行执行，使其运行结果与串行代码一致。
本章不会重复讲解线程创建和 join() 的基础知识，而是聚焦在四种锁上 —— Lock、RLock、Semaphore、Condition。读完本章，你应该能根据业务场景选对锁，并且能一眼看出别人的代码哪里该加锁、哪里锁多了。
threading.Lock 是一个独占锁，同一时间只有一个线程能持有它。如果其他线程试图 acquire（加锁），会被阻塞直到锁被释放。
先来看一个经典竞争条件（Race Condition）：
import threading
import time
counter = 0
def increment():
global counter
for _ in range(100000):
# 读取 -> 修改 -> 写入，这三步不是原子的
current = counter
time.sleep(0.0000001) # 强制让出 CPU，放大竞争概率
counter = current + 1
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"预期值: 1000000, 实际值: {counter}")运行后结果大概率会小于 1000000。原因是 counter += 1 在字节码层面被拆成了多条指令，线程 A 读到 5 还没写回，线程 B 也读到 5，最终只加了 1 次。
在多线程基础章节我们介绍了通过 lock.acquire() 和 lock.release() 实现加锁和释放锁。下面示例演示通过 with lock 进行加锁/释放锁：
import threading
# 创建锁
lock = threading.Lock()
counter = 0
def safe_increment():
global counter
for _ in range(100000):
# 使用锁来保护对共享资源的访问
with lock:
counter += 1
threads = [threading.Thread(target=safe_increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Result: {counter}") # 稳定输出 1000000注意，with lock: 是 lock.acquire() 和 lock.release() 的语法糖，即使在临界区抛出异常，锁也能保证被释放。如果你手写 try...finally，代码会冗长不少：
lock.acquire()
try:
counter += 1
finally:
lock.release()如果锁的范围越大，线程并行度越低（例如直接将整个方法锁住，那么整个方法一次只能被一个线程执行，这将很大程度降低线程的并行执行度）。如果范围太小，又容易遗漏。看一个转账的例子：
import threading
class Account:
def __init__(self, balance):
self.balance = balance
self.lock = threading.Lock() # 每个 Account 实例一把锁
def bad_transfer(a: Account, b: Account, amount):
"""错误：分别加锁，中间状态可能被其他线程读取"""
with a.lock:
a.balance -= amount
with b.lock:
b.balance += amount
def good_transfer(a: Account, b: Account, amount):
"""正确：对涉及的所有账户同时加锁，原子完成"""
# 底粒度锁，并发现更好
# 按 id 排序加锁，避免死锁
first, second = (a, b) if id(a) < id(b) else (b, a)
# 同时持有转账两个账户的锁才能进行转账操作
with first.lock:
with second.lock:
a.balance -= amount
b.balance += amount
alice = Account(1000)
bob = Account(1000)
for _ in range(100):
t1 = threading.Thread(target=good_transfer, args=(alice, bob, 1))
t2 = threading.Thread(target=good_transfer, args=(bob, alice, 1))
t1.start(); t2.start()
t1.join(); t2.join()
print(f"Alice: {alice.balance}, Bob: {bob.balance}") # 都是 1000查看上面代码，你可能存在如下疑问：
为什么要使用 first, second = (a, b) if id(a) < id(b) else (b, a) 来确定锁的顺序。这样写 不行吗？
def good_transfer2(a: Account, b: Account, amount):
"""正确：对涉及的所有账户同时加锁，原子完成"""
with a.lock:
with b.lock:
a.balance -= amount
b.balance += amount如果能保证两个账户之间的转账是串行（执行账户 1 到账户 2，在执行账户 2 到账户 1，...），一笔完成后才会进行下一笔（在正式环境不可能），没有该代码也不会出现问题。如果不能保证，假如同一时刻有两个线程同时执行账户 1 和账户 2 之间的转账，就会出现问题，如下：
线程1：账户1 → 账户2
持有账户1的锁，等待账户2的锁...
线程2：账户2 → 账户1
持有账户2的锁，等待账户1的锁...这时，死锁出现了，线程 1 等待账户 2 的锁，但被线程 2 持有了；线程 2 等待账户 1 的锁，被线程 1 持有了，互补谦让，一直等待下去。
你可能对 good_transfer() 函数有点不理解，为什么不是这样？
# 粗粒度锁，并发性低
GLOBAL_LOCK = Lock()
def transfer(a, b, amount):
with GLOBAL_LOCK:
a.balance -= amount
b.balance += amount这种全局单一锁确实非常简单，但是不推荐使用。为什么？整个系统同一时间只能执行一笔转账。账户 1 转账户 2、账户 3 转账户 4，完全不相关，也要排队串行；用户多、转账频繁时，并发直接卡死，性能极低。
为了提升并发能力，我们为每个 Account 实例自带一把专属锁，只锁住当前操作的两个账户，无关账户互不干扰。
T1：账户 A ↔ B 转账，占用 A、B 锁
T2：账户 C ↔ D 转账，占用 C、D 锁
两者无资源交集，可以同时并行执行，并发能力大幅提升。
注意了：
（1）锁必须覆盖整个事务：读取、计算、写入缺一不可。
（2）多个锁按固定顺序获取：不按顺序加锁，两个线程互相等待对方释放，就会形成死锁。
RLock（Reentrant Lock）允许同一个线程多次 acquire（上锁），内部维护一个持有计数，只有计数归零时才真正释放。它解决的是“锁内调用锁”的问题。
假设你有一个线程安全的类，每个方法都加了锁。方法 A 调用方法 B，如果用的是普通 Lock，就会把自己锁死：
import threading
class BadCounter:
def __init__(self):
self.lock = threading.Lock() # 持有的锁
self.value = 0
def add(self, n):
# 加锁
with self.lock:
self.value += n
def double_add(self, n):
# 加锁
with self.lock:
self.add(n) # 再次请求同一把锁，死锁！
self.add(n)
counter = BadCounter()
def worker():
try:
print("worker() 开始执行...")
counter.double_add(5) # 死锁，永远不会执行后续的代码了
print("worker() 结束执行")
except Exception as e:
print(f"Error: {e}")
# 实际上这里会永远阻塞，不是抛出异常，因为 acquire 是阻塞的
t = threading.Thread(target=worker)
t.start()
t.join()此时，我们可以把 Lock 换成 RLock 就能解决：
import threading
class SafeCounter:
def __init__(self):
self.lock = threading.RLock() # 换成可重入锁
self.value = 0
def add(self, n):
# 加锁
with self.lock:
self.value += n
def double_add(self, n):
# 加锁
with self.lock:
self.add(n) # 同一线程再次 acquire，计数 +1
self.add(n)
counter = SafeCounter()
def worker():
counter.double_add(5)
print(f"Value: {counter.value}")
t = threading.Thread(target=worker)
t.start()
t.join() # 正常结束，输出 10另一个常见场景是递归算法需要线程安全：
import threading
# 创建可重入锁 RLock
rlock = threading.RLock()
def traverse(node, depth=0):
"""递归遍历二叉树，累加所有节点值，使用RLock保证线程安全"""
# 同一线程递归再次进入with会重复acquire，RLock靠计数不会阻塞
with rlock:
# 当前节点为空，递归终止，返回0不参与求和
if node is None:
return 0
# 递归遍历左子树：再次调用traverse，会再次执行with rlock加锁
# 因为是同一个线程持有锁，RLock计数+1，不会死锁
left = traverse(node.get('left'), depth + 1)
# 递归遍历右子树，同理重复上锁
right = traverse(node.get('right'), depth + 1)
# 当前节点值 + 左子树总和 + 右子树总和，向上返回
return node.get('value', 0) + left + right
# 模拟二叉树字典结构
tree = {
'value': 1,
'left': {'value': 2, 'left': {'value': 4}, 'right': {'value': 5}},
'right': {'value': 3, 'left': {'value': 6}, 'right': {'value': 7}}
}
result = traverse(tree)
print(f"Tree sum: {result}") # 输出 28RLock 的持有者信息是线程绑定的。如果线程 A 加了锁，线程 B 试图释放，会抛出 RuntimeError。这个设计是对的，否则锁的语义就崩塌了。例如：
import threading
rlock = threading.RLock()
def thread_a():
# 线程A加锁
rlock.acquire()
print("A acquired")
def thread_b():
# 线程B释放锁
try:
rlock.release() # RuntimeError: cannot release un-acquired lock
except RuntimeError as e:
print(f"B failed: {e}")
t1 = threading.Thread(target=thread_a)
t1.start()
t1.join()
t2 = threading.Thread(target=thread_b)
t2.start()
t2.join()
# 运行结果：
# A acquired
# B failed: cannot release un-acquired lock记住了，RLock 只能在同一个线程中进行加锁和释放锁，不要跨线程。
Semaphore 可以看成一把有 N 个钥匙的锁。最多允许 N 个线程同时进入临界区，第 N+1 个线程必须等待有人退出。它的典型用途是限流，比如限制同时下载的线程数、限制数据库连接池的使用数量。
用法很简单，和 lock 类似，创建一个信号量，然后通过 with 进行使用，例如：
import threading
import time
import random
# 创建信号量
# 最多允许 3 个线程同时下载
sem = threading.Semaphore(3)
def download(url: str):
# 信号量使用
with sem:
print(f"[{threading.current_thread().name}] 开始下载 {url}")
time.sleep(random.uniform(1, 3)) # 模拟下载
print(f"[{threading.current_thread().name}] 完成下载 {url}")
# 准备10个URL
urls = [f"https://example.com/file{i}.zip" for i in range(10)]
# 创建10个线程去下载，一个URL一个线程
threads = [
threading.Thread(target=download, args=(url,), name=f"Worker-{i}")
for i, url in enumerate(urls)
]
for t in threads: t.start()
for t in threads: t.join()
print("全部任务完成")运行后你会看到最多只有 3 个 "开始下载" 同时打印，其他线程在排队。这与线程池不同：Semaphore 不管理线程生命周期，只限制同时执行的代码段数量。如果你需要复用线程，应该用 concurrent.futures.ThreadPoolExecutor(max_workers=3)。
Semaphore 也常用来实现简单的连接池或插槽池，例如：
import threading
import time
class ConnectionPool:
def __init__(self, max_connections=5):
self._semaphore = threading.Semaphore(max_connections)
self._available = max_connections
self._lock = threading.Lock()
def acquire(self):
"""加锁"""
self._semaphore.acquire()
with self._lock:
self._available -= 1
print(f"连接已获取，剩余可用: {self._available}")
return self # 简化返回 self，实际应返回连接对象
def release(self):
"""释放锁"""
with self._lock:
self._available += 1
print(f"连接已释放，剩余可用: {self._available}")
self._semaphore.release()
# 进入 with 代码块，自动执行 __enter__()
def __enter__(self):
self.acquire()
return self
# 离开 with 代码块（正常走完 / 报错异常），自动执行 __exit__()
def __exit__(self, *args):
self.release()
pool = ConnectionPool(3)
def worker(name):
with pool:
print(f"[{name}] 使用连接中...")
time.sleep(1)
print(pool)
print(f"[{name}] 使用完毕")
for i in range(5):
t = threading.Thread(target=worker, args=(f"Thread-{i}",))
t.start()注意，上面示例使用了 with 进入和退出的魔法函数来实现获取连接和释放连接的操作。
标准 Semaphore 有一个隐患：release() 可以被无限制调用，导致计数超过初始值。如果代码里有 bug 多调了几次 release，信号量就形同虚设了。例如：
sem = Semaphore(2)
sem.acquire() # count=1
sem.release() # count=2
sem.release() # 多调用一次！count=3上述代码，现在计数器变成 3，超过初始化的 2。再来 3 个线程 acquire() 全部能拿到许可，原本限制最多 2 人，现在变成 3 人同时执行，限流逻辑直接失效。
BoundedSemaphore 会在 release() 导致计数超过初始值时抛出 ValueError，帮助你尽早发现错误：
import threading
sem = threading.BoundedSemaphore(2)
sem.acquire()
sem.acquire()
sem.release()
sem.release()
sem.release() # ValueError: Semaphore released too many times注意：在需要严格配额控制的场景下，优先使用 BoundedSemaphore。
Condition 是锁的增强版，它内部持有一把锁（默认 RLock），并额外提供 wait()、notify() 和 notify_all() 三个方法。核心场景是生产者-消费者：消费者发现队列空了就等待，生产者放入数据后通知消费者醒来。
下面使用 Condition 来实现生产者-消费者模型，队列大小为 3，如下：
import threading
import time
import random
class Queue:
def __init__(self, maxsize=10):
# 最大大小
self.maxsize = maxsize
# 数据
self.items = []
# 条件变量
self.cond = threading.Condition()
def put(self, item):
"""放入数据到队列"""
with self.cond:
# 队列满时等待，直到有空位
while len(self.items) >= self.maxsize:
print(f"[Producer] 队列已满，等待...")
self.cond.wait() # 等待队列被消费
self.items.append(item)
print(f"[Producer] 放入 {item}，队列: {self.items}")
self.cond.notify_all() # 通知可能在等待的消费者
def get(self):
"""从队列取值"""
with self.cond:
# 队列空时等待，直到有数据
while len(self.items) == 0:
print(f"[Consumer] 队列已空，等待...")
self.cond.wait() # 等待队列被放入数据
item = self.items.pop(0)
print(f"[Consumer] 取出 {item}，队列: {self.items}")
self.cond.notify_all() # 通知可能在等待的生产者
return item
# 创建队列，最大大小为3
queue = Queue(maxsize=3)
def producer():
for i in range(10):
time.sleep(random.uniform(0.1, 0.5))
queue.put(f"item-{i}") # 放值到队列
def consumer():
for _ in range(10):
time.sleep(random.uniform(0.3, 0.7))
queue.get() # 从队列取值
# 生产者
p = threading.Thread(target=producer, name="Producer")
# 消费者
c = threading.Thread(target=consumer, name="Consumer")
p.start()
c.start()
p.join()
c.join() # 主线程等待生产者和消费者完成
print("Done")注意了：
wait() 必须在持有锁的情况下调用，调用后它会自动释放锁并阻塞线程，被唤醒后自动重新获取锁。
用 while 而不是 if 判断条件。因为 notify() 唤醒后，条件可能又不满足了（虚假唤醒或被其他线程抢先），必须再次检查。
notify() 唤醒一个等待线程，notify_all() 唤醒所有。如果不确定该唤醒几个，用 notify_all() 更安全。
wait() 函数支持超时，避免永久阻塞，例如：
with cond:
if not cond.wait(timeout=5.0): # 5 秒内未被唤醒，返回 False
print("等待超时，继续处理备选逻辑")这比用 time.sleep() 做轮询优雅得多。轮询浪费 CPU，而 wait() 会让线程进入阻塞状态，不占用调度资源。
Python 标准库的 queue.Queue 内部已经用 Condition 实现了线程安全的生产者-消费者模型。实际项目中，优先用 queue.Queue，而不是自己手写 Condition。
自己写 Condition 适合以下情况：
队列逻辑有特殊判定规则（如"当队列中总金额超过 1000 时才通知处理"）；
需要多个条件变量协同（如"队列为空且系统未暂停"）。
下表从语义、使用场景和注意点简单说说上面介绍的各种锁的差异：
| 锁类型 | 核心语义 | 适用场景 | 注意点 |
| Lock | 独占，一个线程持有 | 单一临界区、计数器、标志位保护 | 不能重入，跨方法调用时小心死锁 |
| RLock | 可重入，同线程多次 acquire | 类内部方法互相调用、递归算法 | 不能跨线程释放，性能略低于 Lock |
| Semaphore | 最多 N 个线程同时进入 | 限流、连接池、资源配额控制 | 用 BoundedSemaphore 防溢出；注意与线程池的区别 |
| Condition | 等待 + 通知 的精确控制 | 生产者-消费者、状态变化事件 | 必须配合 while 使用；wait() 自动释放/获取锁 |
锁不是万能药。Python 的 GIL（Global Interpreter Lock）已经保护了字节码执行的原子性，对于单条字节码操作（如列表的 append、字典的 pop）实际上不需要额外加锁。但 GIL 会在 I/O 操作和固定数量的字节码指令后释放，所以跨多条语句的复合操作仍然需要锁。
另一个方向是无锁编程。对于简单的计数器，可以用 queue.Queue 做任务分发，或者把状态收敛到单线程（如 asyncio 的事件循环）。锁越少，死锁概率越低，调试越轻松。
死锁是多线程编程的噩梦，而且死锁通常需要在特定条件才会触发，这就给发现死锁增加了难度。死锁也很危险，代码上线后直接死锁，导致功能不可用，甚至整个服务都会崩溃。后果还是很严重，在测试环境不一定能够发现。
线程 1 拿 A 等 B，线程 2 拿 B 等 A，循环等待，死锁发生。例如：
import threading
lock_a = threading.Lock()
lock_b = threading.Lock()
def thread_1():
with lock_a:
print("Thread-1 got A")
threading.Event().wait(0.1) # 故意让出时间片
with lock_b: # 等待锁B
print("Thread-1 got B")
def thread_2():
with lock_b:
print("Thread-2 got B")
threading.Event().wait(0.1)
with lock_a: # 等待锁A
print("Thread-2 got A")
t1 = threading.Thread(target=thread_1)
t2 = threading.Thread(target=thread_2)
t1.start()
t2.start()
# 程序永远卡在这里
t1.join()
t2.join()你可能会问，有没有什么办法避免死锁，答案是肯定的。但是不能完全避免，但是可以降低死锁出现的概率。
一定要记牢了：
（1）全局排序：所有锁按固定顺序获取（比如按 id(lock) 从小到大）。前面转账例子已经展示。
（2）超时机制：lock.acquire(timeout=5.0) 在 5 秒后放弃，打印日志并回滚。
（3）减少锁的持有时间：不要在锁里调外部接口、读写文件，只保留最小必要操作。
（4）能用一个锁，不用两个：锁的数量和死锁概率成正比。
锁的获取和释放需要操作系统内核介入（futex 系统调用），通常耗时在微秒级别。对于高并发、细粒度操作，锁竞争会成为瓶颈。此时考虑：
把数据分区（每个线程操作自己的分片，最后合并）。
使用 threading.local() 做线程本地存储，避免共享。
在合适的场景下用 multiprocessing（多进程处理，后续介绍）绕过 GIL。
调试多线程时，threading.enumerate() 和 threading.current_thread() 很有用：
import threading
def show_threads():
for t in threading.enumerate():
print(f" {t.name} (daemon={t.daemon}, alive={t.is_alive()})")
# 在关键位置插入调用，观察线程是否卡在锁上更高级的调试可以用 faulthandler 模块在程序崩溃时 dump 所有线程的 traceback，或者直接用 py-spy / gdb attach 到进程查看锁的持有者。
注意了，多线程代码的正确性不能靠 "看起来没问题" 来保证。写出一段并发代码只是开始，在压力测试下跑 10 万次、100 万次、... 不崩溃，才算真正过关。
川公网安备51010802032098