本文目录导读：
我来为你介绍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
关键点总结：
- 使用Lock保护共享资源：使用
with lock: - 避免死锁：固定锁获取顺序或使用超时机制
- 考虑性能：锁的粒度要合适，过大的锁范围会影响性能
- 使用RLock：需要重入的场景使用可重入锁
- 使用高级同步原语：Condition、Semaphore等适合特定场景
