没吃饱的笨蛋

外观

线程安全与锁

约 1701 字大约 6 分钟

2026-05-10

阅读导览
线程安全与锁

多个线程同时访问同一份可变数据时，程序结果可能变得不可预测。这类问题叫线程安全问题。

  • 竞态条件
  • Lock
  • acquire 和 release
  • RLock
实践清单
学习练习清单
  1. 1跑两个线程各 counter += 1 十万次，看到结果不到 200000 —— 这就是 += 不是原子操作的证据。
  2. 2用 with lock: 包住临界区；不要 lock.acquire() 没 release()，异常会让锁永远不释放。
  3. 3把读余额 + 扣余额放进同一把锁里 —— 检查和修改之间漏锁就是 TOCTOU 漏洞。
  4. 4故意写两个线程交叉拿两把锁，看死锁现象；改成“固定加锁顺序”后即可避免。
  5. 5锁里别做网络/磁盘 IO；范围越小越好，能用 Queue 传数据就别共享变量。

多个线程同时访问同一份可变数据时，程序结果可能变得不可预测。这类问题叫线程安全问题。

竞态条件

竞态条件（Race Condition）指的是：程序结果取决于多个线程执行的先后顺序。

import threading

counter = 0

def add():
    global counter
    for _ in range(100000):
        counter += 1

threads = [threading.Thread(target=add) for _ in range(2)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print(counter)

理论上结果应该是 200000，但在复杂情况下，counter += 1 并不是一个不可分割的操作，它大致包含读取、加一、写回三个步骤。如果多个线程交错执行，就可能丢失更新。

Lock

threading.Lock 用来保护临界区。临界区指的是同一时间只能允许一个线程执行的代码区域。

import threading

counter = 0
lock = threading.Lock()

def add():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

with lock: 会自动获取锁和释放锁。即使中间发生异常，锁也能被释放。

acquire 和 release

也可以手动获取和释放锁：

lock.acquire()
try:
    counter += 1
finally:
    lock.release()

但更推荐使用 with lock:，因为它更安全。

RLock

普通 Lock 不能被同一个线程重复获取。RLock 是可重入锁，允许同一个线程多次获取同一把锁。

import threading

lock = threading.RLock()

def outer():
    with lock:
        inner()

def inner():
    with lock:
        print('重复获取同一把锁')

RLock 适合方法之间互相调用、但都需要保护同一份资源的场景。

死锁

死锁指的是多个线程互相等待对方释放锁，导致程序永远无法继续。

# 线程 A：先拿 lock1，再等 lock2
# 线程 B：先拿 lock2，再等 lock1

避免死锁的常见方法：

  • 固定加锁顺序
  • 缩小锁的范围
  • 不要在持锁时执行耗时 IO
  • 必要时使用超时参数
  • 尽量减少共享状态

不要在锁里做耗时操作

反例：

with lock:
    data.append(item)
    send_to_remote_server(item)

网络请求很慢，会导致其他线程长时间等待锁。

更好的做法：

with lock:
    data.append(item)

send_to_remote_server(item)

锁只保护必须保护的共享数据，不要把无关操作放进去。

减少共享状态

线程安全最好的办法不是“到处加锁”，而是减少共享数据。

常见策略：

  • 每个线程只处理自己的局部变量
  • 使用 queue.Queue 在线程之间传递数据
  • 使用不可变对象
  • 把共享状态集中到少数对象中管理

账户示例

import threading

class Account:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance < amount:
                raise ValueError('余额不足')
            self.balance -= amount

account = Account(100)

余额检查和扣款必须放在同一把锁里，否则两个线程可能同时看到余额足够，然后都扣款成功。

注意事项

  1. 锁可以保证同一时间只有一个线程进入临界区。
  2. 锁的范围越小越好，但不能小到破坏原子性。
  3. 不要在锁里做网络请求、文件 IO、长时间计算。
  4. 多把锁要规定固定获取顺序。
  5. 优先用队列传递数据，而不是让多个线程直接修改共享变量。

为什么看起来简单的 += 也不安全

很多人以为：

counter += 1

是一瞬间完成的。实际上它大致可以拆成几步：

读取 counter 当前值
加 1
把新值写回 counter

如果两个线程同时做这件事，就可能出现：

线程 A 读到 10
线程 B 也读到 10
线程 A 写回 11
线程 B 也写回 11

明明加了两次，结果只增加了 1。这就是竞态条件。

锁保护的是一段临界区

锁不是保护变量本身，而是保护“同一时间只能有一个线程执行的那段代码”。

import threading

counter = 0
lock = threading.Lock()


def add_one():
    global counter
    with lock:
        counter += 1

with lock: 里面的代码越短越好。只把真正需要互斥的部分放进去，不要把网络请求、文件读写、长时间计算都塞进锁里。

用 with 比手动 acquire/release 更稳

手动写：

lock.acquire()
# do something
lock.release()

如果中间报错，release() 可能执行不到，锁就永远不释放。

更推荐：

with lock:
    # do something

即使中间抛异常，Python 也会帮你释放锁。

减少共享，比到处加锁更好

线程安全最好的办法不一定是“给所有地方都加锁”，而是减少共享数据。

更推荐的思路：

  • 每个线程处理自己的局部数据；
  • 线程之间用 queue.Queue 传递消息；
  • 汇总结果时集中处理；
  • 不让多个线程随意修改同一个全局变量。

加锁可以解决问题，但也会让程序更难理解。共享越少，锁越少，bug 也越少。

死锁的日常版解释

死锁可以理解成两个人互相等对方让路：

线程 A 拿着锁 1，等待锁 2
线程 B 拿着锁 2，等待锁 1

两边都不放手，程序就卡住了。

避免死锁的简单建议：

  1. 尽量少同时持有多把锁；
  2. 如果必须拿多把锁，固定加锁顺序；
  3. 锁里不要调用复杂函数；
  4. 锁里不要做网络请求或长时间等待。

总结

线程安全问题来自共享可变状态。锁可以解决竞态条件，但也会带来死锁、性能下降和复杂性。写多线程程序时，应优先减少共享状态，其次才是正确加锁。

重点总结
要点回收
  • 竞态条件来自多个线程同时读写共享状态，`counter += 1` 也可能不安全。
  • 锁保护的是临界区，范围越小越好；优先使用 `with lock:`。
  • 减少共享数据通常比到处加锁更可靠，线程间通信优先考虑 Queue。

版权归属：Shuo Liu

许可证：署名-非商业性 4.0 国际 (CC-BY-NC-4.0)