从单标志到互斥锁用Python代码复现操作系统中的经典互斥算法在并发编程的世界里多个线程或进程同时访问共享资源时如何确保数据一致性和系统稳定性是一个永恒的话题。想象一下当两个线程同时试图修改同一个银行账户余额时如果没有适当的同步机制结果可能会让你大吃一惊——而这正是互斥算法要解决的核心问题。本文将带你用Python代码一步步复现操作系统教科书中的经典互斥算法从最基础的单标志法到现代编程中广泛使用的互斥锁让你在动手实践中真正理解并发控制的精髓。1. 环境准备与基础概念在开始编码之前我们需要明确几个关键术语临界区(Critical Section)访问共享资源的代码段同一时刻只允许一个线程执行竞态条件(Race Condition)多个线程无序访问共享数据导致的结果不确定性互斥(Mutual Exclusion)确保任何时候只有一个线程能进入临界区我们将使用Python的threading模块创建多个线程来模拟并发场景。每个算法实现都会包含以下基本结构import threading shared_resource 0 # 被多个线程共享的变量 def thread_function(): global shared_resource # 进入区代码 # 临界区代码 # 退出区代码 threads [] for i in range(2): t threading.Thread(targetthread_function) threads.append(t) t.start() for t in threads: t.join() print(Final value:, shared_resource)理想情况下如果两个线程各对shared_resource加1操作1000次最终值应该是2000。但如果缺少互斥保护实际结果通常会小于这个值——这就是我们要解决的问题。2. 软件实现的互斥算法2.1 单标志法最简单的轮流机制单标志法通过一个共享变量turn来决定哪个线程可以进入临界区turn 0 # 0表示第一个线程可以进入1表示第二个线程可以进入 def thread_0(): global turn, shared_resource for _ in range(1000): while turn ! 0: # 如果不是我的回合就等待 pass shared_resource 1 # 临界区 turn 1 # 交出权限 def thread_1(): global turn, shared_resource for _ in range(1000): while turn ! 1: # 如果不是我的回合就等待 pass shared_resource 1 # 临界区 turn 0 # 交出权限问题与观察强制线程交替执行即使一个线程不想进入临界区另一个线程也必须等待违反空闲让进原则当临界区空闲时想进入的线程可能因为turn值不对而无法进入运行程序会发现结果总是正确的2000但性能极差——线程必须严格轮流执行2.2 双标志先检查法表达进入意愿改进思路是让每个线程独立表达自己是否想进入临界区flag [False, False] # 每个线程有自己的标志位 def thread_0(): global flag, shared_resource for _ in range(1000): while flag[1]: # 先检查对方是否想进入 pass flag[0] True # 表达自己想进入 shared_resource 1 # 临界区 flag[0] False # 退出临界区 def thread_1(): global flag, shared_resource for _ in range(1000): while flag[0]: # 先检查对方是否想进入 pass flag[1] True # 表达自己想进入 shared_resource 1 # 临界区 flag[1] False # 退出临界区问题与观察可能出现两个线程同时检查对方标志发现都为False然后同时进入临界区运行多次会发现最终结果有时小于2000证明互斥失败违反忙则等待原则无法保证同时只有一个线程在临界区2.3 双标志后检查法先声明再检查调整标志设置的顺序先表明自己的意图再检查对方flag [False, False] def thread_0(): global flag, shared_resource for _ in range(1000): flag[0] True # 先声明自己想进入 while flag[1]: # 再检查对方 pass shared_resource 1 flag[0] False def thread_1(): global flag, shared_resource for _ in range(1000): flag[1] True while flag[0]: pass shared_resource 1 flag[1] False问题与观察可能出现两个线程都先设置自己的标志然后互相等待对方导致死锁程序可能会卡住无法继续执行违反空闲让进和有限等待原则2.4 Peterson算法优雅的谦让机制结合前几种方法的优点Peterson算法引入了谦让机制flag [False, False] turn 0 def thread_0(): global flag, turn, shared_resource for _ in range(1000): flag[0] True # 我想进入 turn 1 # 但你可以先进入 while flag[1] and turn 1: # 如果你也想进入且轮到你了 pass shared_resource 1 flag[0] False def thread_1(): global flag, turn, shared_resource for _ in range(1000): flag[1] True turn 0 while flag[0] and turn 0: pass shared_resource 1 flag[1] False优势与局限总能保证互斥和有限等待但仍然存在忙等待(自旋)浪费CPU资源代码复杂度相对较高运行结果始终正确的20003. 硬件辅助的互斥实现3.1 模拟Test-And-Set指令现代CPU通常提供原子指令来实现互斥。我们可以用Python的ctypes库模拟import ctypes def test_and_set(p): old p.value p.value True return old lock ctypes.c_bool(False) def thread_0(): global lock, shared_resource for _ in range(1000): while test_and_set(lock): # 获取锁 pass shared_resource 1 lock.value False # 释放锁 def thread_1(): global lock, shared_resource for _ in range(1000): while test_and_set(lock): pass shared_resource 1 lock.value False特点分析测试和设置操作是原子的不会被中断解决了软件方法的所有问题除了让权等待适用于多处理器系统仍然是忙等待可能浪费CPU周期3.2 使用真正的原子操作Python的threading模块提供了基于原子操作的锁lock threading.Lock() def thread_0(): global lock, shared_resource for _ in range(1000): lock.acquire() # 原子操作 shared_resource 1 lock.release() def thread_1(): global lock, shared_resource for _ in range(1000): lock.acquire() shared_resource 1 lock.release()性能对比方法正确性忙等待实现复杂度适用场景单标志法✓✓低教学演示双标志先检查✗✓低不推荐实际使用双标志后检查✗✓低不推荐实际使用Peterson算法✓✓中教学和研究Test-And-Set✓✓中底层系统实现互斥锁✓✗低实际生产环境首选4. 现代互斥锁的深入应用Python的标准库提供了几种锁的实现各有特点4.1 基本互斥锁lock threading.Lock() def safe_increment(): global shared_resource for _ in range(1000): with lock: # 上下文管理器自动获取和释放锁 shared_resource 14.2 可重入锁(RLock)允许同一个线程多次获取锁rlock threading.RLock() def recursive_func(n): if n 0: return with rlock: recursive_func(n-1)4.3 条件变量(Condition)实现更复杂的同步逻辑condition threading.Condition() queue [] def producer(): global queue for i in range(10): with condition: queue.append(i) condition.notify() # 通知等待的消费者 def consumer(): global queue while True: with condition: while not queue: condition.wait() # 等待通知 item queue.pop(0) print(Consumed:, item)4.4 信号量(Semaphore)控制同时访问资源的线程数量semaphore threading.Semaphore(3) # 最多3个线程同时访问 def limited_resource_access(): with semaphore: print(Accessing limited resource) time.sleep(1)在实际项目中选择哪种同步机制取决于具体需求。互斥锁是最基础也最常用的选择而条件变量和信号量则适用于更复杂的同步场景。