Python 编程：从零基础到精通

第 50 课 - 线程安全与锁（Lock, RLock）

所属模块：模块十一：并发与并行难度：高级标签：concurrency, thread-safety, locks 上一课：并发编程：多线程（threading）入门 下一课：并行计算：多进程（multiprocessing）

1. 学习目标

完成本课学习后，你将能够：

理解竞态条件（Race Condition）：解释为什么多个线程同时修改共享数据会导致不可预测的结果。
掌握锁（Lock）的基本使用：使用 threading.Lock 来保护临界区代码，确保线程安全。
理解可重入锁（RLock）的用途：区分 Lock 和 RLock，并在需要锁嵌套的场景中正确使用 RLock。
了解线程安全的数据结构：知道 queue.Queue 等模块提供了现成的线程安全容器，减少手动加锁的需求。

2. 核心概念

在上一课中，我们学习了如何创建和启动线程。当多个线程开始同时运行时，一个潜在的问题就会浮出水面：共享资源。

想象一下，两个线程（两个柜员）同时操作同一个银行账户（共享资源）进行取款操作。

线程A读取余额为1000元。
线程B也读取余额为1000元。
线程A计算 1000 - 300 = 700，并将余额更新为700。
线程B计算 1000 - 500 = 500，并将余额更新为500。

最终余额是500元，但正确的结果应该是200元（1000 - 300 - 500）。这种由于线程执行顺序的不确定性而导致的程序错误，就叫做竞态条件（Race Condition）。账户余额这个共享数据，因为多个线程无序地"竞争"访问，而陷入了不一致的状态。

如何解决？ 我们需要一种机制，确保当一个线程在修改共享数据时，其他线程不能同时进行修改。这就像给房间（临界区）上一把锁（Lock）。一个线程进入房间后，从里面把门锁上，其他线程必须在门外等待，直到里面的线程出来并开锁。

Lock（锁）：最基本的同步原语。它有两种状态："锁定"和"非锁定"。一个线程可以通过 acquire() 方法获取锁（如果锁已被其他线程持有，则会阻塞等待），并通过 release() 方法释放锁。为了简化使用，我们通常使用 with lock: 语句，它会自动获取和释放锁。
RLock（可重入锁）：普通锁有一个问题：如果一个已经获取了锁的线程尝试再次获取同一个锁，它会阻塞（因为锁被自己持有，但它在等自己释放，形成死锁）。RLock 允许同一个线程多次获取同一把锁，而不会导致死锁。它内部维护一个计数器，每获取一次加1，每释放一次减1，只有当计数器归零时，锁才真正被释放。这对于锁嵌套（一个持有锁的函数调用了另一个也需要同一把锁的函数）的场景至关重要。

3. 代码示例

示例1：竞态条件演示与Lock的修复

import threading
import time

# 共享资源
balance = 1000
lock = threading.Lock()  # 创建一把锁

def withdraw_unsafe(amount):
    """不安全的取款函数，演示竞态条件"""
    global balance
    print(f"开始取款 {amount} 元，当前余额: {balance}")
    # 模拟读取、计算、写入的操作序列
    current_balance = balance
    time.sleep(0.001)  # 故意加入一点延迟，放大竞态条件发生的概率
    new_balance = current_balance - amount
    if new_balance >= 0:
        balance = new_balance
        print(f"成功取出 {amount} 元，新余额: {balance}")
    else:
        print(f"余额不足，取款失败。当前余额: {balance}")

def withdraw_safe(amount):
    """使用Lock的安全取款函数"""
    global balance
    # 使用 with 语句自动管理锁的获取和释放
    with lock:
        print(f"开始取款 {amount} 元，当前余额: {balance}")
        current_balance = balance
        time.sleep(0.001)
        new_balance = current_balance - amount
        if new_balance >= 0:
            balance = new_balance
            print(f"成功取出 {amount} 元，新余额: {balance}")
        else:
            print(f"余额不足，取款失败。当前余额: {balance}")

# 创建两个线程进行取款
print("=== 测试不安全的函数（可能出现错误）===")
t1 = threading.Thread(target=withdraw_unsafe, args=(300,))
t2 = threading.Thread(target=withdraw_unsafe, args=(500,))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"最终余额: {balance}\n")  # 可能看到错误结果

# 重置余额，测试安全的函数
balance = 1000
print("=== 测试安全的函数（使用Lock）===")
t3 = threading.Thread(target=withdraw_safe, args=(300,))
t4 = threading.Thread(target=withdraw_safe, args=(500,))
t3.start()
t4.start()
t3.join()
t4.join()
print(f"最终余额: {balance}")  # 保证结果正确，应为200

示例2：RLock的应用场景

当需要嵌套使用锁时，Lock 会导致死锁，而 RLock 则不会。

import threading

rlock = threading.RLock()  # 创建一把可重入锁

def recursive_function(count):
    """一个需要嵌套调用自身（并需要访问共享资源）的函数"""
    with rlock:  # 第一次获取锁
        print(f"递归层: {count}")
        if count > 0:
            # 递归调用，函数内部会再次尝试获取同一把锁
            recursive_function(count - 1)  # 第二、三...次获取锁
        # 离开 with 代码块时，锁的计数减一。只有最外层退出时才真正释放。

def outer_task():
    """演示任务"""
    with rlock:
        print("外层任务开始")
        recursive_function(3)  # 嵌套调用
        print("外层任务结束")

# 启动一个线程运行任务
thread = threading.Thread(target=outer_task)
thread.start()
thread.join()
# 如果使用 Lock，程序会在 recursive_function 中死锁。
# 使用 RLock，程序能正常运行。

示例3：线程安全的Queue（队列）

在生产者-消费者模型中，queue.Queue 是比手动管理锁更高级、更安全的选择。它内部已经处理好了所有的同步问题。

import threading
import queue
import time
import random

# 创建一个线程安全的队列
task_queue = queue.Queue()

def producer():
    """生产者：生成任务放入队列"""
    for i in range(5):
        task = f"任务-{i}"
        task_queue.put(task)  # 线程安全的入队操作
        print(f"生产者: 放入 {task}")
        time.sleep(random.uniform(0.1, 0.5))
    # 放入一个特殊值表示结束
    task_queue.put(None)

def consumer():
    """消费者：从队列中取出任务处理"""
    while True:
        task = task_queue.get()  # 线程安全的出队操作（队列为空时会阻塞）
        if task is None:  # 收到结束信号
            break
        print(f"消费者: 处理 {task}")
        time.sleep(random.uniform(0.2, 0.4))
        task_queue.task_done()  # 标记任务完成

# 创建并启动生产者和消费者线程
prod_thread = threading.Thread(target=producer)
cons_thread = threading.Thread(target=consumer)
prod_thread.start()
cons_thread.start()
prod_thread.join()
cons_thread.join()
print("所有任务处理完毕。")

4. 实践练习

练习1：修复计数器 有一个全局变量 count = 0，10个线程每个线程都将其增加1000次。请使用 Lock 修复以下代码，确保最终结果总是10000。

# 有缺陷的代码框架
import threading

count = 0
# 你的锁和函数代码...

def increment():
    global count
    for _ in range(1000):
        count += 1  # 这行代码不是线程安全的！

threads = [threading.Thread(target=increment) for _ in range(10)]
# ... 启动和等待线程
print(f"最终计数: {count}") # 期望：10000

练习2：设计一个线程安全的银行账户类 设计一个 BankAccount 类，包含 deposit(存款) 和 withdraw(取款) 方法。该类内部应使用锁来确保余额 balance 的操作是线程安全的。考虑使用 Lock 还是 RLock？为什么？

练习3（挑战）：性能对比 创建一个共享的列表，让多个线程同时向其末尾添加元素。

方案A：使用 Lock 保护列表的 append 操作。
方案B：使用线程安全的 queue.Queue，最后将队列中的元素取出放入列表。比较两种方案的代码复杂度和结果正确性。（提示：list.append() 在CPython中通常是线程安全的，但依赖于此是不好的实践。本练习意在练习锁的使用。）

5. 常见错误

忘记释放锁：如果使用 lock.acquire() 获取锁，但在所有路径上没有调用 lock.release()（例如在 try...except 块中），会导致其他线程永远阻塞。最佳实践是始终使用 with lock: 语句。
在持有锁时进行耗时操作：如果一个线程持有锁的时间过长（例如在里面进行IO操作或复杂的计算），其他需要这把锁的线程将不得不长时间等待，这会严重降低程序的并发性能。锁的临界区应尽可能短小。
在锁的范围内调用未知代码：如果在 with lock: 块中调用了一个第三方函数或用户传入的回调函数，你不知道这个函数内部会做什么。如果它也尝试获取同一把锁，可能会造成死锁（如果锁是普通的Lock）。
过度使用锁：为所有共享变量都上锁会使程序退化成串行执行，并增加死锁风险。首先考虑是否真的需要共享该数据，或者是否可以使用线程安全的数据结构（如 Queue）来替代。
混淆Lock和RLock的使用场景：仅在确实需要可重入特性时才使用 RLock，因为它的开销比 Lock 略大。在大多数简单场景下，Lock 就足够了。

6. 小结

本课我们深入探讨了多线程编程中核心的挑战——线程安全：

竞态条件是并发编程的"魔鬼"，它源于对共享数据的无序访问。
锁（Lock） 是对抗竞态条件的基础武器，它能确保临界区的互斥访问。使用 with lock: 是管理锁生命周期的最佳方式。
可重入锁（RLock） 解决了锁的嵌套获取问题，避免了在同一线程内因重复获取同一把锁而导致的死锁。
线程安全容器（如 queue.Queue）提供了更高层次的抽象，在很多场景下可以让我们更安全、更简洁地编写并发代码，而无需手动管理锁。

记住，锁是用来保护数据，而不是代码。正确、精细地使用锁，是编写健壮的多线程应用程序的关键。下一课，我们将学习另一种实现并发的方式——多进程，它能更好地利用多核CPU进行并行计算。

练习编辑器

python