Python 多线程 threading：竞态条件与锁
Python 多线程 threading：理解竞态条件与锁
在多线程编程的世界里，提高程序执行效率的同时，也引入了一个核心挑战：线程安全。本文将带你深入 Python threading 模块，通过生动的示例理解什么是竞态条件，以及如何使用锁（Lock）来保护共享资源，写出正确、可靠的多线程代码。即使是初学者，也能轻松掌握这些关键概念。
为什么需要线程同步？
当我们创建多个线程去执行同一个任务时，它们通常需要访问和修改共享的数据结构，比如全局变量、列表或字典。问题在于，线程的调度由操作系统决定，任何一个线程都可能在任何时刻被暂停，转而执行另一个线程。如果多个线程不加控制地同时读写同一份数据，最终的结果可能变得不可预测，这就是**竞态条件（Race Condition）**的根源。
举个例子：你和小伙伴共用一张银行卡，两人同时在不同的 ATM 上取款，如果不检查余额并加锁，很可能取出超过总额的钱。在 Python 中，我们需要一种机制来确保一次只有一个线程能够操作关键数据。
复现竞态条件：一个银行账户的例子
让我们从一个简单的 BankAccount 类开始，它包含存款和取款方法。我们创建 100 个线程同时执行 1000 次存款操作，每次存入 1 元。理论上，最终余额应该是 100×1000 = 100000 元，但如果没有同步，结果会出错。
import threading
class BankAccount:
def __init__(self):
self.balance = 0
def deposit(self, amount):
# 读取当前余额
current = self.balance
# 模拟线程切换，放大竞态窗口
for _ in range(1000):
pass
# 写入新余额
self.balance = current + amount
def make_deposits(account, count):
for _ in range(count):
account.deposit(1)
account = BankAccount()
threads = []
for _ in range(100):
t = threading.Thread(target=make_deposits, args=(account, 1000))
threads.append(t)
t.start()
for t in threads:
t.join()
print(f"最终余额: {account.balance}") # 几乎不会是 100000
代码解释
deposit方法不是原子操作：它分三步执行——读、计算、写。线程可能在读出current后、写回self.balance前被中断。for _ in range(1000): pass故意延长了计算间隔，让竞态条件更容易发生。- 运行几次后你会发现，余额总是小于 100000，甚至每次结果都不同。这就是**丢失更新（Lost Update）**的典型表现。
用锁（Lock）保护临界区
解决竞态条件的标准方案是使用互斥锁（Mutex）。Python threading 模块提供了 Lock 对象，它能确保一段代码（称为临界区）在同一时刻只能由一个线程执行。锁有两种状态：锁定（locked）和未锁定（unlocked）。线程通过 acquire() 获取锁，用 release() 释放锁。如果锁已被其他线程持有，当前线程会阻塞，直到锁被释放。
我们将上面的例子改造为线程安全的版本。
import threading
class SafeBankAccount:
def __init__(self):
self.balance = 0
self.lock = threading.Lock()
def deposit(self, amount):
with self.lock: # 获取锁，离开 with 块时自动释放
current = self.balance
for _ in range(1000):
pass
self.balance = current + amount
def make_deposits(account, count):
for _ in range(count):
account.deposit(1)
account = SafeBankAccount()
threads = []
for _ in range(100):
t = threading.Thread(target=make_deposits, args=(account, 1000))
threads.append(t)
t.start()
for t in threads:
t.join()
print(f"最终余额: {account.balance}") # 稳定输出 100000
使用 with 语句管理锁
推荐使用上下文管理器（with lock:）来替代手动调用 acquire() 和 release()。它的优点在于：
- 自动释放锁，哪怕临界区内发生异常，也能避免死锁。
- 代码更简洁，不易忘记释放。
等价于：
self.lock.acquire()
try:
# 临界区代码
pass
finally:
self.lock.release()
锁的工作原理与注意事项
锁的核心行为
- 当一个线程调用
acquire()时，如果锁未被锁定，线程立即获得锁并继续执行；如果锁已被占用，线程会进入阻塞状态，直到锁变为可用。 release()只能由持有锁的线程调用，否则会引发RuntimeError。- 在
with语句块内部，你可以放心地执行读取、计算、写入等非原子操作。
性能与死锁的权衡
锁虽然保证了正确性，但也带来了性能开销和潜在的死锁风险。
- 性能：当一个线程持有锁时，其他试图进入同一临界区的线程会被阻塞，这实际上将并发执行变为了串行执行。因此，临界区的代码应该尽可能短，避免包含耗时的 I/O 或复杂计算。
- 死锁：当两个或多个线程互相等待对方释放锁时，就会发生死锁。典型场景：
预防死锁的简单原则：所有线程获取多个锁时，必须按照相同的顺序。lock_a = threading.Lock() lock_b = threading.Lock() def thread1(): with lock_a: with lock_b: # do work pass def thread2(): with lock_b: # 注意获取顺序与 thread1 相反 with lock_a: # do work pass
锁的变体：RLock（可重入锁）
threading.RLock（Reentrant Lock，重入锁）允许同一个线程多次获取同一个锁，而不会死锁。它内部维护着一个计数器和所属线程的记录。典型场景是递归函数或当一个线程调用的方法需要获取它已经持有的锁时。
lock = threading.RLock()
def recursive_function(n):
with lock:
if n > 0:
recursive_function(n-1) # 可以再次获取同一个锁
在普通 Lock 中，同一个线程尝试第二次 acquire() 会陷入永久阻塞。日常使用中，除非你明确需要可重入特性，否则简单的 Lock 已经足够。
实际应用中的最佳实践
-
最小化临界区范围
只保护那些必须互斥访问的代码行，不要让大段计算、I/O 或网络请求待在with lock:块里。例如：# 不好的做法 with lock: data = fetch_from_db() # 耗时 I/O process(data) # 更好的做法 data = fetch_from_db() with lock: update_shared_resource(data) -
使用单个锁保护多个相关资源
如果多个变量必须作为一个整体被原子性地更新，那么它们应该被同一个锁保护。 -
避免在持有锁时抛出未捕获的异常
虽然with语句会在异常时自动释放锁，但最好在临界区内编写健壮的代码，防止资源处于不一致状态。 -
测试线程安全
竞态条件可能很难被检测出来。重复运行测试（例如使用pytest-timeout和大量线程），或借助threading.Barrier等同步原语强制线程在特定点会合，可以帮助暴露问题。
进阶知识预览
掌握了基本的 Lock 后，你还可以探索 threading 模块中的其他同步工具：
- Semaphore：允许同时有固定数量的线程访问资源。
- Event：一个线程发出事件信号，其他线程等待该信号。
- Condition：更复杂的等待/通知机制，常用于生产者-消费者模型。
- Queue：线程安全的队列，本身就是生产者-消费者模型的优秀实现，内部已处理好锁。
不过，在深入这些之前，请务必把“竞态条件与锁”这块基石踩实。它们是构建一切可靠多线程程序的基础。
小结
- 竞态条件因多线程非同步访问共享数据而产生，导致计算结果错误。
- **锁（Lock）**通过互斥访问临界区，保证了操作的原子性，是解决竞态条件的核心工具。
- 使用时请遵循临界区最小化、固定锁获取顺序等原则，避免性能问题和死锁。
- Python 的
with语句让锁管理既安全又优雅。
现在，你已经可以自信地在自己的 Python 项目中运用线程锁，写出既快又稳的并发代码。打开编辑器，亲自试一试之前的银行账户示例，观察加锁前后的区别吧！