Python多历程同步全分析：从竞争条件到锁、信号量的实战应用

1. 历程同步的须要性

在多历程编程中，当多个历程须要访问共享资源时，会出现竞争条件标题。比方火车票售卖体系中，如果多个售票窗口同时读取和修改剩余票数，大概导致数据差别等。
1.1 竞争条件示例

2. from multiprocessing import Process, Value
4. def sell_tickets(tickets):
5.     for _ in range(1000):
6.         tickets.value -= 1  # 非原子操作
8. if __name__ == '__main__':
9.     tickets = Value('i', 10000)
10.     processes = [Process(target=sell_tickets, args=(tickets,))
11.                 for _ in range(10)]
12.    
13.     for p in processes:
14.         p.start()
15.     for p in processes:
16.         p.join()
17.    
18.     print(f"剩余票数: {tickets.value}")  # 结果不确定，可能大于0

复制代码

在多历程环境下，多个历程可以同时实行。像代码里的tickets.value -= 1利用，现实上并非原子利用。它包罗读取tickets.value的值、对该值减 1、再把新值写回tickets.value这几个步调。当多个历程同时实行这一利用时，就大概出现数据差别等的状态。好比，两个历程同时读取到剩余票数为 100，各自减 1 后都写回 99，而现实上应该是 98。
2. 锁(Lock)机制

锁机制可以包管同一时间只有一个历程可以或许访问共享资源。当一个历程获取到锁之后，其他历程就必须等候该历程开释锁，才气继承访问共享资源。如许就能制止多个历程同时修改共享资源，从而办理竞争条件标题。
2.1 底子锁使用

2. from multiprocessing import Process, Lock, Value
4. def sell_tickets(tickets, lock):
5.     for _ in range(1000):
6.         lock.acquire()
7.         tickets.value -= 1
8.         lock.release()
10. if __name__ == '__main__':
11.     tickets = Value('i', 10000)
12.     lock = Lock()
13.     processes = [Process(target=sell_tickets, args=(tickets, lock))
14.                 for _ in range(10)]
15.    
16.     for p in processes:
17.         p.start()
18.     for p in processes:
19.         p.join()
20.    
21.     print(f"剩余票数: {tickets.value}")

复制代码

在上述代码中，lock.acquire()用于获取锁，一旦某个历程获取到锁，其他历程就只能等候。tickets.value -= 1是对共享资源的修改利用，由于有锁的掩护，同一时间只有一个历程能实行这一利用。lock.release()用于开释锁，开释后其他历程就可以竞争获取锁。
2.2 上下文管理器简化

2. from multiprocessing import Process, Lock, Value
4. def sell_tickets(tickets, lock):
5.     for _ in range(1000):
6.         with lock:  # 自动获取和释放锁
7.             tickets.value -= 1
9. if __name__ == '__main__':
10.     tickets = Value('i', 10000)
11.     lock = Lock()
12.     processes = [Process(target=sell_tickets, args=(tickets, lock))
13.                 for _ in range(10)]
14.    
15.     for p in processes:
16.         p.start()
17.     for p in processes:
18.         p.join()
19.    
20.     print(f"剩余票数: {tickets.value}")

复制代码

使用with lock语句，Python 会自动帮我们获取和开释锁。进入with语句块时，会自动调用lock.acquire()获取锁；脱离with语句块时，会自动调用lock.release()开释锁。如允许以让代码更加轻巧，也能制止忘记开释锁的标题。
2.3 银行转账案例

原始标题代码（存在竞争条件）

2. from multiprocessing import Process, Value
3. import time
5. def transfer(balance, amount):
6.     # 没有锁保护的转账操作
7.     current = balance.value
8.     time.sleep(0.01)  # 模拟处理延迟
9.     balance.value = current + amount  # 非原子操作
11. if __name__ == '__main__':
12.     balance = Value('i', 100)  # 初始余额100
13.    
14.     # 创建100个存款和100个取款进程
15.     processes = []
16.     for _ in range(100):
17.         p_deposit = Process(target=transfer, args=(balance, 1))  # 存1元
18.         p_withdraw = Process(target=transfer, args=(balance, -1)) # 取1元
19.         processes.extend([p_deposit, p_withdraw])
20.    
21.     for p in processes:
22.         p.start()
23.     for p in processes:
24.         p.join()
25.    
26.     print(f"最终余额: {balance.value}")  # 预期应该是100，但实际结果不确定

复制代码

标题分析：

•         多个历程同时读取和修改balance.value会导致竞争条件
  
•         balance.value = current + amount不是原子利用
  
•         由于time.sleep放大了竞争窗口，标题更容易显现
  
•         终极余额大概不即是初始值100，阐明出现了数据差别等
  

改进后的办理方案（使用Lock）

2. from multiprocessing import Process, Value, Lock
3. import time
5. def transfer(balance, lock, amount):
6.     with lock:  # 使用上下文管理器自动获取和释放锁
7.         current = balance.value
8.         time.sleep(0.01)  # 模拟处理延迟
9.         balance.value = current + amount
11. if __name__ == '__main__':
12.     balance = Value('i', 100)  # 初始余额100
13.     lock = Lock()  # 创建锁对象
14.    
15.     processes = []
16.     for _ in range(100):
17.         p_deposit = Process(target=transfer, args=(balance, lock, 1))  # 存1元
18.         p_withdraw = Process(target=transfer, args=(balance, lock, -1)) # 取1元
19.         processes.extend([p_deposit, p_withdraw])
20.    
21.     for p in processes:
22.         p.start()
23.     for p in processes:
24.         p.join()
25.    
26.     print(f"最终余额: {balance.value}")  # 现在总能正确输出100

复制代码

改进点阐明：

•         引入Lock对象包管对共享变量的互斥访问
  
•         使用with lock:上下文管理器简化锁的获取和开释
  
•         确保读取-修改-写入利用成为原子利用
  
•         纵然有耽误，也能包管数据同等性
  

实行效果对比

版本运行次数大概效果阐明原始版本10次90-110之间随机因竞争条件效果不确定改进版本10次每次都是100包管数据同等性可视化标题发生过程

2. 时间线       进程A(存款)               进程B(取款)              余额值
3. --------------------------------------------------------------------
4. t1      读取balance=100
5. t2                          读取balance=100
6. t3      计算100+1=101
7. t4                          计算100-1=99
8. t5      写入101
9. t6                          写入99

复制代码

由于没有同步机制，后写入的利用会覆盖前一个利用，导致存款利用"丢失"。
更优实践发起

对于这种计数器场景，可以使用Value的原子利用：

2. from multiprocessing import Value
4. balance = Value('i', 100)
5. balance.value += 1  # 非原子操作（不安全）
6. balance.get_lock().acquire()
7. balance.value += 1  # 原子操作（安全）
8. balance.get_lock().release()
10. # 或者使用with语句
11. with balance.get_lock():
12.     balance.value += 1

复制代码

如允许以直接使用共享变量自带的锁，代码更轻巧。
3. 信号量(Semaphore)

信号量是更高级的同步原语，答应指定命目的历程同时访问资源。
3.1 底子信号量

2. from multiprocessing import Semaphore
4. s = Semaphore(3)  # 允许3个进程同时访问
6. s.acquire()
7. print("进程1进入")
8. s.acquire()
9. print("进程2进入")
10. s.acquire()
11. print("进程3进入")
12. # 此时第4个进程会被阻塞

复制代码

3.2 课堂钥匙案例

2. from multiprocessing import Process, Semaphore
3. import time
4. import random
6. def student(id, sem):
7.     with sem:
8.         print(f"学生{id}进入教室")
9.         time.sleep(random.uniform(1, 3))
10.         print(f"学生{id}离开教室")
12. if __name__ == '__main__':
13.     sem = Semaphore(5)  # 5把钥匙
14.     students = [Process(target=student, args=(i, sem))
15.                for i in range(20)]
16.    
17.     for s in students:
18.         s.start()
19.     for s in students:
20.         s.join()

复制代码

4. 变乱(Event)机制

变乱用于历程间的简朴通讯，一个历程发出"变乱已发生"信号，其他历程等候该变乱。
4.1 变乱底子

2. from multiprocessing import Event
4. e = Event()
5. print(e.is_set())  # False
7. e.set()
8. print(e.is_set())  # True
9. e.wait()  # 立即通过
11. e.clear()
12. e.wait()  # 阻塞直到set()被调用

复制代码

4.2 红绿灯控制案例

2. from multiprocessing import Process, Event
3. import time
5. def traffic_light(e):
6.     while True:
7.         print("\033[32m绿灯亮\033[0m" if e.is_set() else "\033[31m红灯亮\033[0m")
8.         time.sleep(2)
9.         e.set() if not e.is_set() else e.clear()
11. def car(e, car_id):
12.     e.wait()
13.     print(f"汽车{car_id}通过路口")
15. if __name__ == '__main__':
16.     e = Event()
17.     light = Process(target=traffic_light, args=(e,))
18.     light.daemon = True  # 守护进程，主进程结束自动退出
19.     light.start()
20.    
21.     for i in range(10):
22.         time.sleep(0.5)
23.         Process(target=car, args=(e, i)).start()

复制代码

5. 综合案例：火车票售卖体系

5.1 共享内存实现

2. from multiprocessing import Process, Value, Lock
3. import time
5. class TicketSeller(Process):
6.     def __init__(self, tickets, lock, name):
7.         super().__init__()
8.         self.tickets = tickets
9.         self.lock = lock
10.         self.name = name
11.         
12.     def run(self):
13.         while True:
14.             with self.lock:
15.                 if self.tickets.value <= 0:
16.                     break
17.                 time.sleep(0.1)  # 模拟处理时间
18.                 print(f"{self.name}售出票号{self.tickets.value}")
19.                 self.tickets.value -= 1
21. if __name__ == '__main__':
22.     tickets = Value('i', 100)
23.     lock = Lock()
24.    
25.     sellers = [TicketSeller(tickets, lock, f"窗口{i}")
26.               for i in range(1, 4)]
27.    
28.     for s in sellers:
29.         s.start()
30.     for s in sellers:
31.         s.join()
32.    
33.     print("所有票已售完")

复制代码

5.2 文件共享实现

2. from multiprocessing import Process, Lock
3. import time
5. def sell_tickets(lock, window):
6.     while True:
7.         with lock:
8.             with open("tickets.txt", "r+") as f:
9.                 tickets = int(f.read())
10.                 if tickets <= 0:
11.                     break
12.                 time.sleep(0.1)
13.                 print(f"{window}售出票号{tickets}")
14.                 f.seek(0)
15.                 f.write(str(tickets-1))
17. if __name__ == '__main__':
18.     lock = Lock()
19.    
20.     # 初始化票数
21.     with open("tickets.txt", "w") as f:
22.         f.write("100")
23.    
24.     processes = [Process(target=sell_tickets, args=(lock, f"窗口{i}"))
25.                 for i in range(1, 4)]
26.    
27.     for p in processes:
28.         p.start()
29.     for p in processes:
30.         p.join()
31.    
32.     print("所有票已售完")

复制代码

6. 同步机制选择指南

机制实用场景优点缺点Lock互斥访问共享资源简朴直接大概引起死锁Semaphore限定资源访问数量机动控制并发度比锁复杂Event历程间简朴关照轻量级关照机制不通报数据7. 常见标题解答

Q1: 锁和信号量的区别是什么？

•         锁是二元信号量(0/1)，信号量可以设置多个允许
  
•         锁用于互斥访问，信号量用于控制资源访问数量
  

Q2: 怎样制止死锁？

•         按固定序次获取多个锁
  
•         使用超机遇制
  
•         制止嵌套锁
  

Q3: 文件共享和共享内存哪个更好？

•         共享内存更快但更复杂
  
•         文件共享更简朴但性能较低
  
•         根据场景选择符合方案