示例工程教程【免费下载链接】materialsBonus materials, exercises, and example projects for our Python tutorials项目地址https://gitcode.com/gh_mirrors/ma/materials点击查看免费下载导读本文以 materials 仓库 thread-safety-locks 目录下的九份代码示例为主线系统讲解 Python 多线程编程中最核心的线程安全问题。文章从竞态条件导致银行账户取款结果不可预测的实例出发逐一剖析threading.Lock互斥锁、RLock可重入锁、Semaphore信号量、Event事件、Condition条件变量与Barrier屏障六种同步原语的用法、底层语义与适用场景并配套ThreadPoolExecutor线程池底座。读完本文你将能识别共享状态上的竞争、用正确的同步原语保护临界区、规避死锁并学会按场景选型。本目录代码源自 Real Python 同名教程Python Thread Safety: Using a Lock and Other Techniques作者 Adarsh Divakaran以 MIT 协议分发详见仓库根目录 LICENSE 与 thread-safety-locks/README.md所有示例均可直接运行验证。一、线程安全问题的根源竞态条件为什么有 GIL还会出问题CPython 解释器通过 GIL全局解释器锁保证同一时刻只有一个线程执行 Python 字节码因此单条字节码指令是原子的。但像读余额 → 计算新余额 → 写回余额这样的read-modify-write操作序列由多条指令构成线程调度可能发生在任意两条指令之间。当多个线程并发执行这类非原子序列并操作同一份共享数据时就产生了竞态条件race condition——最终结果取决于线程调度顺序无法预测、不可复现。不安全的银行账户bank_multithreaded_withdrawal.py仓库中的 bank_multithreaded_withdrawal.py 用最朴素的方式演示了这一缺陷import time from concurrent.futures import ThreadPoolExecutor class BankAccount: def __init__(self): self.balance 1000 def withdraw(self, amount): if self.balance amount: new_balance self.balance - amount time.sleep(0.1) # Simulate a delay self.balance new_balance else: raise ValueError(Insufficient balance) account BankAccount() with ThreadPoolExecutor(max_workers2) as executor: executor.submit(account.withdraw, 500) executor.submit(account.withdraw, 700) print(fFinal account balance: {account.balance})初始余额为 1000两个线程分别尝试取款 500 与 700。单独看任意一笔取款都是合法的1000 ≥ 500、1000 ≥ 700两次都成功后的正确余额应为 500。但withdraw()中检查余额 → 计算new_balance→sleep(0.1)模拟耗时 → 写回self.balance是多步操作两个线程完全可能在对方写回之前都通过余额检查两个线程都读取到balance 1000各自计算出500与300后写回者覆盖前者余额变成300 或 500更糟的情况是某个线程在另一个线程写回后才读取余额得到300 - 500 -200账户被透支由于两次取款都在检查点通过甚至不会抛出ValueError(Insufficient balance)。用python thread-safety-locks/bank_multithreaded_withdrawal.py多运行几次会观察到最终余额在不同数值间摇摆。这正是共享可变状态上竞态条件的典型表现。二、threading.Lock用互斥锁把临界区变成原子操作加锁后的线程安全版本bank_thread_safe.pybank_thread_safe.py 在账户内新增一把threading.Lock()并把检查 计算 写回整体放进with临界区import threading import time from concurrent.futures import ThreadPoolExecutor class BankAccount: def __init__(self, balance0): self.balance balance self.account_lock threading.Lock() def withdraw(self, amount): with self.account_lock: if self.balance amount: new_balance self.balance - amount print(fWithdrawing {amount}...) time.sleep(0.1) # Simulate a delay self.balance new_balance else: raise ValueError(Insufficient balance) def deposit(self, amount): with self.account_lock: new_balance self.balance amount print(fDepositing {amount}...) time.sleep(0.1) # Simulate a delay self.balance new_balance account BankAccount(1000) with ThreadPoolExecutor(max_workers3) as executor: executor.submit(account.withdraw, 700) executor.submit(account.deposit, 1000) executor.submit(account.withdraw, 300) print(fFinal account balance: {account.balance})由于Lock是互斥的任意时刻只有一个线程能进入with self.account_lock保护的临界区withdraw()与deposit()内部的检查-计算-写回整体成为原子操作。三个任务并发执行时无论调度顺序如何最终余额恒为1000余额不足时也能按预期抛出ValueError。sleep(0.1)不再引发错误因为睡眠时锁依然被持有其他线程只能排队等待。Lock 的核心语义APIacquire()获取锁默认阻塞、release()释放锁、locked()查询状态with语句会在进入时获取、退出时自动释放即使临界区抛出异常也能保证释放是首选用法阻塞行为acquire(blockingTrue)会一直等待也可传timeout参数做限时等待并返回布尔结果非重入同一线程对已持有的普通Lock再次acquire()会永久阻塞——这正是下一节死锁的来源性能取舍锁让并发执行退化为串行因此临界区要尽量短小只保护真正需要原子性的读改写序列如bank_thread_safe.py中balance的读写不要把无关的耗时操作圈进来。三、死锁陷阱与 threading.RLock可重入锁普通 Lock 引发的自我死锁bank_deadlock.py当加锁方法内部再调用另一个同样需要加锁的方法时普通Lock会出问题。bank_deadlock.py 给出了一个会挂死的版本class BankAccount: def __init__(self): self.balance 0 self.lock threading.Lock() def deposit(self, amount): with self.lock: time.sleep(0.1) self._update_balance(amount) def _update_balance(self, amount): with self.lock: # This will cause a deadlock self.balance amount执行流程deposit()获取self.lock后调用_update_balance()后者再次对同一线程已持有的同一把锁执行acquire()。普通Lock不记录持有者身份这次acquire永远不会成功线程在_update_balance()的with处永久阻塞。三个Worker线程全部卡死程序挂起、不再打印最终余额只能靠CtrlC中断。这种由同一线程重复获取非重入锁导致的死锁常被称为重入死锁self-deadlock是嵌套调用场景下最常见的并发陷阱之一。用 RLock 解决重入bank_rlock.pybank_rlock.py 与死锁版本结构完全一致唯一区别是把锁换成可重入锁self.lock threading.RLock()RLock内部跟踪持有线程身份 递归计数同一线程可以多次acquire()同一把RLock每次acquire必须有对应的releasewith嵌套即自动匹配其他线程仍然只能在该线程完全释放后才获准进入互斥性不变。因此三个线程各执行deposit(100)都能顺利完成最终余额恒为300程序正常退出。适用场景包括一个锁保护的多层方法调用链如deposit() → _update_balance()、递归函数中需要反复进入同一临界区以及先加锁再调用本类其他加锁方法的代码结构。四、threading.Semaphore限制并发资源访问的信号量Lock保证同一时刻只有一个线程当资源允许多个但有限线程同时访问时应使用Semaphore。bank_semaphore.py 用银行柜员场景演示银行只有 2 个柜员席位最多同时服务 2 位顾客。# Semaphore with a maximum of 2 resources (tellers) teller_semaphore threading.Semaphore(2) def serve_customer(name): print(f{now()}: {name} is waiting for a teller.) with teller_semaphore: print(f{now()}: {name} is being served by a teller.) # Simulate the time taken for the teller to serve the customer time.sleep(random.randint(1, 3)) print(f{now()}: {name} is done being served.)5 位顾客通过ThreadPoolExecutor(max_workers5)同时提交但Semaphore(2)内部维护一个初始值为 2 的计数器进入with时acquire()使计数减 1减到 0 时后续线程阻塞退出时release()使计数加 1唤醒等待者。运行时会看到任意时刻最多 2 位顾客处于 being served 状态其余在排队等待。Semaphore的典型用途数据库连接池、限流限制并发请求数、控制同时执行的任务数量。值得了解的是threading.BoundedSemaphore——它在release()次数超过初始值时会抛出ValueError能及早暴露多释放类逻辑错误比普通Semaphore更适合作为资源池。五、threading.Event线程间的事件通知Event解决的是线程等待某个外部条件发生的同步问题它本质上是一个线程可设置、可查询的标志位。bank_event.py 用银行开门、开放交易两个阶段控制顾客的行为bank_open threading.Event() transactions_open threading.Event() def serve_customer(customer_data): print(f{customer_data[name]} is waiting for the bank to open.) bank_open.wait() print(f{customer_data[name]} entered the bank) if customer_data[type] WITHDRAW_MONEY: print(f{customer_data[name]} is waiting for transactions to open.) transactions_open.wait() print(f{customer_data[name]} is starting their transaction.) # Simulate the time taken for performing the transaction time.sleep(2) print(f{customer_data[name]} completed transaction and exited bank) else: # Simulate the time taken for banking time.sleep(2) print(f{customer_data[name]} has exited bank)主线程在提交 4 位顾客任务后按营业节奏依次触发事件print(Bank is now open!) bank_open.set() # Signal that the bank is open time.sleep(3) print(Transactions are now open!) transactions_open.set()4 位顾客中3 位WITHDRAW_MONEY型顾客要等待两个事件都set()后才开始取款1 位CHECK_BALANCE型顾客只需等银行开门。Event的 API 语义set()置位标志所有wait()的线程被唤醒wait(timeoutNone)阻塞直到标志被置位或超时返回Falseclear()复位标志后续wait()再次阻塞is_set()查询当前状态。Event非常适合一次性放行如启动信号或闸门式阶段同步如果需求是某个条件反复出现/反复等待则应当改用Condition。六、threading.Condition条件变量与生产者-消费者模型Condition是更通用的同步原语它内部组合了一把锁并提供等待与通知机制是生产者-消费者模型的经典实现工具。bank_condition.py 中柜员线程是消费者顾客线程是生产者customer_available_condition threading.Condition() # Customers waiting to be served by the Teller customer_queue [] def serve_customers(): while True: with customer_available_condition: # Wait for a customer to arrive while not customer_queue: print(f{now()}: Teller is waiting for a customer.) customer_available_condition.wait() # Serve the customer customer customer_queue.pop(0) print(f{now()}: Teller is serving {customer}.) # Simulate the time taken to serve the customer time.sleep(random.randint(1, 5)) print(f{now()}: Teller has finished serving {customer}.) def add_customer_to_queue(name): with customer_available_condition: print(f{now()}: {name} has arrived at the bank.) customer_queue.append(name) customer_available_condition.notify()运行机制与要点wait()会释放关联锁并阻塞当前线程等被唤醒后重新获取锁再继续执行——这保证了条件检查与队列操作的一致性notify()唤醒一个正在wait()的线程notify_all()唤醒全部等待者调用二者前必须先持有条件锁消费者侧用while not customer_queue:而不是if检查条件目的是防御虚假唤醒与多消费者竞争——被唤醒后可能队列又被其他消费者清空主线程每隔random.randint(1, 3)秒提交一位顾客共 5 位顾客入队后立即notify()柜员随即出队服务。运行注意serve_customers()是while True无限循环脚本会持续运行不退出演示时需手动中断CtrlC生产代码应配合退出信号例如额外设置一个Event作为停止服务标记优雅关闭而不是依赖中断。七、threading.Barrier多线程同步就绪屏障Barrier屏障用于让n 个线程都到达某个点后再一起放行。bank_barrier.py 中三位柜员各自准备柜台耗时 13 秒不等全部准备完毕才允许开始服务顾客teller_barrier threading.Barrier(3) def prepare_for_work(name): print(f{now()}: {name} is preparing their counter.) # Simulate the delay to prepare the counter time.sleep(random.randint(1, 3)) print(f{now()}: {name} has finished preparing.) # Wait for all tellers to finish preparing teller_barrier.wait() print(f{now()}: {name} is now ready to serve customers.)Barrier(3)要求恰好 3 个线程调用wait()先到达的线程阻塞直到最后一个线程到达屏障才一次性放行全部线程。因此无论各柜员准备耗时如何随机三条is now ready to serve customers一定在同一时刻之后紧挨着出现而主线程的All tellers are ready一定在所有柜员放行完毕之后打印。Barrier的适用场景并行阶段计算如全部线程完成第一阶段后再进入第二阶段、批量初始化、回合制同步。它默认是可复用的——所有线程放行后屏障自动重置可再次用于下一轮同步。八、并发底座ThreadPoolExecutor 与线程命名上述所有示例都使用标准库concurrent.futures.ThreadPoolExecutor提交并发任务它屏蔽了手工创建/管理线程的细节。threading_example.py 单独演示了线程池与线程命名def threaded_function(): for number in range(3): print(fPrinting from {threading.current_thread().name}. {number}) time.sleep(0.1) with ThreadPoolExecutor( max_workers4, thread_name_prefixWorker ) as executor: for _ in range(4): executor.submit(threaded_function)要点max_workers4限定线程池最多同时运行 4 个线程thread_name_prefixWorker为池内线程统一命名如Worker_0、Worker_1配合threading.current_thread().name可以清晰区分每行输出来自哪个线程executor.submit(func, *args)异步提交任务并返回Future对象线程池作为上下文管理器退出时会调用shutdown(waitTrue)等待所有已提交任务完成后才继续执行后面的代码各银行示例如bank_thread_safe.py中的三任务、bank_semaphore.py中的五任务均复用这一底座运行方式统一为python thread-safety-locks/文件名.py观察输出即可验证同步效果提示{number}是 Python 3.8 引入的 f-string 调试语法同时打印变量名与值本目录示例因此要求Python 3.8 及以上版本。九、总结同步原语选型对照与最佳实践选型对照表同步原语解决的问题核心 API仓库示例Lock互斥保护临界区消除竞态条件acquire()/release()/withbank_thread_safe.pyRLock同一线程可重入的互斥锁同Lock支持嵌套获取bank_rlock.pySemaphore限制同时访问某资源的线程数acquire()/release()bank_semaphore.pyEvent线程间事件通知标志位set()/clear()/wait()/is_set()bank_event.pyCondition条件等待 生产者-消费者wait()/notify()/notify_all()bank_condition.pyBarriern 个线程同步到达后再放行wait()bank_barrier.py最佳实践清单最小化共享状态优先设计为各线程持有局部/不可变数据把并发范围收窄到必须共享的部分用with语句管理锁进入即获取、退出即释放异常也不会遗留未释放的锁控制锁粒度临界区只覆盖真正需要原子的读改写序列过粗降低并发度过细则易引入死锁与一致性问题多锁嵌套保持固定获取顺序多个线程以相同顺序获取多把锁避免循环等待经典死锁的四个必要条件之一同一线程内嵌套加锁用RLock如方法调用链跨多个加锁函数见 bank_rlock.py按语义选型互斥用Lock/RLock限流用Semaphore通知用Event条件队列用Condition回合同步用Barrier验证方法对线程安全版本如bank_thread_safe.py多次运行最终余额应保持稳定一致对非安全版本bank_multithreaded_withdrawal.py则会出现结果漂移——这正是判断代码是否存在竞态条件的直观手段。掌握以上六种同步原语并结合ThreadPoolExecutor规范提交任务就能在 Python 中写出行为可预测、结果可复现的线程安全代码。仓库内 thread-safety-locks 目录提供了全部可运行示例读者可以逐一运行、对比输出作为并发编程的练手与参考素材。赞分享示例工程教程【免费下载链接】materialsBonus materials, exercises, and example projects for our Python tutorials项目地址https://gitcode.com/gh_mirrors/ma/materials点击查看免费下载相关推荐CPython asyncio 同步原语全解析Lock、Event、Condition、Semaphore 与 Barrier 的用法与源码实现CPython asyncio 同步原语全解析Lock、Event、Condition、Semaphore 与 Barrier 的用法与源码实现 本指南以 C编程语言语言运行时解释器标准库Tornado 协程同步原语完全指南locks 模块的 Event、Condition、Semaphore 与 Lock 深度解析Tornado 协程同步原语完全指南locks 模块的 Event、Condition、Semaphore 与 Lock 深度解析 tornado.locks后端Web框架异步编程WebSocket仓颉 std.sync 并发接口深度解析Condition、Lock、UniqueLock 与 IReentrantMutex 完全指南仓颉 std.sync 并发接口深度解析Condition、Lock、UniqueLock 与 IReentrantMutex 完全指南 导读 本文以仓颉编程编程语言语言运行时标准库垃圾回收并发编程上一篇揭秘gh_mirrors/ke/kernel如何用Rust打造极简OS内核下一篇kubectl-neat核心算法揭秘如何智能识别和移除默认值与冗余信息创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
