Python线程锁案例如何解决线程冲突

本文目录导读：

1. 问题场景：银行转账
2. 线程锁解决方案
3. 死锁场景与解决方案
4. 使用 RLock（可重入锁）
5. 使用条件变量（Condition）
6. 最佳实践

我来为你介绍Python线程锁解决线程冲突的完整案例：

问题场景：银行转账

不安全的版本（无锁）

import threading
import time
class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance
    def transfer(self, amount, from_account, to_account):
        print(f"线程{threading.current_thread().name}: 转账{amount}元")
        # 模拟网络延迟
        time.sleep(0.1)
        # 检查余额
        if from_account.balance >= amount:
            from_account.balance -= amount
            # 模拟网络延迟
            time.sleep(0.1)
            to_account.balance += amount
            print(f"线程{threading.current_thread().name}: 转账成功")
        else:
            print(f"线程{threading.current_thread().name}: 余额不足")
# 创建账户
acc1 = BankAccount(1000)
acc2 = BankAccount(0)
# 创建线程进行并发转账
threads = []
for i in range(10):
    t = threading.Thread(target=lambda: acc1.transfer(100, acc1, acc2))
    threads.append(t)
    t.start()
# 等待所有线程完成
for t in threads:
    t.join()
print(f"最终余额 - 账户1: {acc1.balance}, 账户2: {acc2.balance}")
print("预期: 账户1=0, 账户2=1000")

线程锁解决方案

使用 Lock 的基本版本

import threading
import time
class SafeBankAccount:
    def __init__(self, balance=0):
        self.balance = balance
        self.lock = threading.Lock()
    def transfer(self, amount, from_account, to_account):
        print(f"线程{threading.current_thread().name}: 尝试转账{amount}元")
        # 使用锁保护临界区
        with self.lock:
            print(f"线程{threading.current_thread().name}: 获得锁")
            # 模拟网络延迟
            time.sleep(0.1)
            # 检查余额
            if from_account.balance >= amount:
                from_account.balance -= amount
                # 模拟网络延迟
                time.sleep(0.1)
                to_account.balance += amount
                print(f"线程{threading.current_thread().name}: 转账成功，释放锁")
            else:
                print(f"线程{threading.current_thread().name}: 余额不足，释放锁")
# 测试安全版本
safe_acc1 = SafeBankAccount(1000)
safe_acc2 = SafeBankAccount(0)
threads = []
for i in range(10):
    t = threading.Thread(target=lambda: safe_acc1.transfer(100, safe_acc1, safe_acc2))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"安全转账 - 最终余额 - 账户1: {safe_acc1.balance}, 账户2: {safe_acc2.balance}")

死锁场景与解决方案

容易产生死锁的版本

import threading
import time
class DeadlockBank:
    def __init__(self, balance=0):
        self.balance = balance
        self.lock = threading.Lock()
    def transfer_deadlock(self, amount, from_account, to_account):
        # 先锁定转出账户
        with from_account.lock:
            print(f"锁定转出账户: {threading.current_thread().name}")
            time.sleep(0.1)
            # 再锁定转入账户
            with to_account.lock:
                print(f"锁定转入账户: {threading.current_thread().name}")
                if from_account.balance >= amount:
                    from_account.balance -= amount
                    to_account.balance += amount
# 可能产生死锁的测试
def test_deadlock():
    acc_a = DeadlockBank(1000)
    acc_b = DeadlockBank(500)
    t1 = threading.Thread(target=lambda: acc_a.transfer_deadlock(100, acc_a, acc_b))
    t2 = threading.Thread(target=lambda: acc_b.transfer_deadlock(100, acc_b, acc_a))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
# test_deadlock()  # 可能会死锁

死锁解决方案：固定锁顺序

import threading
import time
class SafeBankAccount2:
    def __init__(self, balance=0, account_id=0):
        self.balance = balance
        self.lock = threading.Lock()
        self.account_id = account_id
    def transfer_safe(self, amount, from_account, to_account):
        # 根据账户ID确定锁的顺序，避免死锁
        first_lock = from_account if from_account.account_id < to_account.account_id else to_account
        second_lock = to_account if first_lock == from_account else from_account
        with first_lock.lock:
            print(f"线程{threading.current_thread().name}: 获得第一个锁")
            time.sleep(0.1)
            with second_lock.lock:
                print(f"线程{threading.current_thread().name}: 获得第二个锁")
                if from_account.balance >= amount:
                    from_account.balance -= amount
                    to_account.balance += amount
                    print(f"线程{threading.current_thread().name}: 转账成功")
                else:
                    print(f"线程{threading.current_thread().name}: 余额不足")
# 测试安全版本
acc_x = SafeBankAccount2(1000, 1)
acc_y = SafeBankAccount2(500, 2)
t1 = threading.Thread(target=lambda: acc_x.transfer_safe(100, acc_x, acc_y))
t2 = threading.Thread(target=lambda: acc_y.transfer_safe(100, acc_y, acc_x))
t1.start()
t2.start()
t1.join()
t2.join()
print(f"安全转账 - 账户X: {acc_x.balance}, 账户Y: {acc_y.balance}")

使用 RLock（可重入锁）

import threading
class RecursiveLockExample:
    def __init__(self):
        self.lock = threading.RLock()
        self.data = 0
    def update(self):
        with self.lock:
            self.data += 1
    def batch_update(self, count):
        with self.lock:  # 同一个线程可以多次获取RLock
            for _ in range(count):
                self.update()  # 内部调用也会获取锁
    def safe_method(self):
        with self.lock:
            self.data += 1
            return self.data
# 使用示例
obj = RecursiveLockExample()
obj.batch_update(5)
print(f"批量更新后数据: {obj.data}")  # 输出: 5

使用条件变量（Condition）

import threading
import time
class ProducerConsumer:
    def __init__(self):
        self.items = []
        self.condition = threading.Condition()
        self.max_size = 5
    def produce(self, item):
        with self.condition:
            while len(self.items) >= self.max_size:
                print("仓库满了，生产者等待...")
                self.condition.wait()
            self.items.append(item)
            print(f"生产: {item}, 库存: {len(self.items)}")
            self.condition.notify()  # 通知消费者
    def consume(self):
        with self.condition:
            while len(self.items) == 0:
                print("仓库空了，消费者等待...")
                self.condition.wait()
            item = self.items.pop(0)
            print(f"消费: {item}, 库存: {len(self.items)}")
            self.condition.notify()  # 通知生产者
            return item
# 测试生产者-消费者模式
pc = ProducerConsumer()
def producer():
    for i in range(10):
        pc.produce(f"商品{i}")
        time.sleep(0.1)
def consumer():
    for i in range(10):
        pc.consume()
        time.sleep(0.2)
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()

最佳实践

import threading
class ThreadSafeCounter:
    def __init__(self):
        self._lock = threading.Lock()
        self._count = 0
    def increment(self):
        with self._lock:
            self._count += 1
    def decrement(self):
        with self._lock:
            self._count -= 1
    @property
    def count(self):
        with self._lock:
            return self._count
# 使用示例
counter = ThreadSafeCounter()
def worker():
    for _ in range(1000):
        counter.increment()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"最终计数: {counter.count}")  # 应该输出 10000

关键点总结：

1. 使用Lock保护共享资源：使用 with lock:
2. 避免死锁：固定锁获取顺序或使用超时机制
3. 考虑性能：锁的粒度要合适，过大的锁范围会影响性能
4. 使用RLock：需要重入的场景使用可重入锁
5. 使用高级同步原语：Condition、Semaphore等适合特定场景