当前位置： 首页 > news >正文

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

news 来源：原创 2025/4/21 6:44:05

1. 进程同步的必要性

在多进程编程中，当多个进程需要访问共享资源时，会出现竞争条件问题。例如火车票售卖系统中，如果多个售票窗口同时读取和修改剩余票数，可能导致数据不一致。

1.1 竞争条件示例

from multiprocessing import Process, Valuedef sell_tickets(tickets):for _ in range(1000):tickets.value -= 1  # 非原子操作if __name__ == '__main__':tickets = Value('i', 10000)processes = [Process(target=sell_tickets, args=(tickets,)) for _ in range(10)]for p in processes:p.start()for p in processes:p.join()print(f"剩余票数: {tickets.value}")  # 结果不确定，可能大于0

在多进程环境下，多个进程可以同时执行。像代码里的tickets.value -= 1操作，实际上并非原子操作。它包含读取tickets.value的值、对该值减 1、再把新值写回tickets.value这几个步骤。当多个进程同时执行这一操作时，就可能出现数据不一致的状况。比如，两个进程同时读取到剩余票数为 100，各自减 1 后都写回 99，而实际上应该是 98。

2. 锁(Lock)机制

锁机制可以保证同一时间只有一个进程能够访问共享资源。当一个进程获取到锁之后，其他进程就必须等待该进程释放锁，才能继续访问共享资源。这样就能避免多个进程同时修改共享资源，从而解决竞争条件问题。

2.1 基础锁使用

from multiprocessing import Process, Lock, Valuedef sell_tickets(tickets, lock):for _ in range(1000):lock.acquire()tickets.value -= 1lock.release()if __name__ == '__main__':tickets = Value('i', 10000)lock = Lock()processes = [Process(target=sell_tickets, args=(tickets, lock)) for _ in range(10)]for p in processes:p.start()for p in processes:p.join()print(f"剩余票数: {tickets.value}")

在上述代码中，lock.acquire()用于获取锁，一旦某个进程获取到锁，其他进程就只能等待。tickets.value -= 1是对共享资源的修改操作，由于有锁的保护，同一时间只有一个进程能执行这一操作。lock.release()用于释放锁，释放后其他进程就可以竞争获取锁。

2.2 上下文管理器简化

from multiprocessing import Process, Lock, Valuedef sell_tickets(tickets, lock):for _ in range(1000):with lock:  # 自动获取和释放锁tickets.value -= 1if __name__ == '__main__':tickets = Value('i', 10000)lock = Lock()processes = [Process(target=sell_tickets, args=(tickets, lock)) for _ in range(10)]for p in processes:p.start()for p in processes:p.join()print(f"剩余票数: {tickets.value}")

使用with lock语句，Python 会自动帮我们获取和释放锁。进入with语句块时，会自动调用lock.acquire()获取锁；离开with语句块时，会自动调用lock.release()释放锁。这样可以让代码更加简洁，也能避免忘记释放锁的问题。

2.3 银行转账案例

原始问题代码（存在竞争条件）
from multiprocessing import Process, Value
import timedef transfer(balance, amount):# 没有锁保护的转账操作current = balance.valuetime.sleep(0.01)  # 模拟处理延迟balance.value = current + amount  # 非原子操作if __name__ == '__main__':balance = Value('i', 100)  # 初始余额100# 创建100个存款和100个取款进程processes = []for _ in range(100):p_deposit = Process(target=transfer, args=(balance, 1))  # 存1元p_withdraw = Process(target=transfer, args=(balance, -1)) # 取1元processes.extend([p_deposit, p_withdraw])for p in processes:p.start()for p in processes:p.join()print(f"最终余额: {balance.value}")  # 预期应该是100，但实际结果不确定

问题分析：

  1. 多个进程同时读取和修改balance.value会导致竞争条件

  2. balance.value = current + amount不是原子操作

  3. 由于time.sleep放大了竞争窗口，问题更容易显现

  4. 最终余额可能不等于初始值100，说明出现了数据不一致

改进后的解决方案（使用Lock）
from multiprocessing import Process, Value, Lock
import timedef transfer(balance, lock, amount):with lock:  # 使用上下文管理器自动获取和释放锁current = balance.valuetime.sleep(0.01)  # 模拟处理延迟balance.value = current + amountif __name__ == '__main__':balance = Value('i', 100)  # 初始余额100lock = Lock()  # 创建锁对象processes = []for _ in range(100):p_deposit = Process(target=transfer, args=(balance, lock, 1))  # 存1元p_withdraw = Process(target=transfer, args=(balance, lock, -1)) # 取1元processes.extend([p_deposit, p_withdraw])for p in processes:p.start()for p in processes:p.join()print(f"最终余额: {balance.value}")  # 现在总能正确输出100

改进点说明：

  1. 引入Lock对象保证对共享变量的互斥访问

  2. 使用with lock:上下文管理器简化锁的获取和释放

  3. 确保读取-修改-写入操作成为原子操作

  4. 即使有延迟，也能保证数据一致性

执行结果对比
版本运行次数可能结果说明
原始版本10次90-110之间随机因竞争条件结果不确定
改进版本10次每次都是100保证数据一致性
可视化问题发生过程
时间线       进程A(存款)               进程B(取款)              余额值
--------------------------------------------------------------------t1      读取balance=100t2                          读取balance=100t3      计算100+1=101t4                          计算100-1=99t5      写入101t6                          写入99

由于没有同步机制，后写入的操作会覆盖前一个操作，导致存款操作"丢失"。

更优实践建议

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

from multiprocessing import Valuebalance = Value('i', 100)
balance.value += 1  # 非原子操作（不安全）
balance.get_lock().acquire()
balance.value += 1  # 原子操作（安全）
balance.get_lock().release()# 或者使用with语句
with balance.get_lock():balance.value += 1

这样可以直接使用共享变量自带的锁，代码更简洁。

3. 信号量(Semaphore)

信号量是更高级的同步原语，允许指定数量的进程同时访问资源。

3.1 基础信号量

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

3.2 教室钥匙案例

from multiprocessing import Process, Semaphore
import time
import randomdef student(id, sem):with sem:print(f"学生{id}进入教室")time.sleep(random.uniform(1, 3))print(f"学生{id}离开教室")if __name__ == '__main__':sem = Semaphore(5)  # 5把钥匙students = [Process(target=student, args=(i, sem)) for i in range(20)]for s in students:s.start()for s in students:s.join()

4. 事件(Event)机制

事件用于进程间的简单通信，一个进程发出"事件已发生"信号，其他进程等待该事件。

4.1 事件基础

from multiprocessing import Evente = Event()
print(e.is_set())  # Falsee.set()
print(e.is_set())  # True
e.wait()  # 立即通过e.clear()
e.wait()  # 阻塞直到set()被调用

4.2 红绿灯控制案例

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

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

5.1 共享内存实现

from multiprocessing import Process, Value, Lock
import timeclass TicketSeller(Process):def __init__(self, tickets, lock, name):super().__init__()self.tickets = ticketsself.lock = lockself.name = namedef run(self):while True:with self.lock:if self.tickets.value <= 0:breaktime.sleep(0.1)  # 模拟处理时间print(f"{self.name}售出票号{self.tickets.value}")self.tickets.value -= 1if __name__ == '__main__':tickets = Value('i', 100)lock = Lock()sellers = [TicketSeller(tickets, lock, f"窗口{i}") for i in range(1, 4)]for s in sellers:s.start()for s in sellers:s.join()print("所有票已售完")

5.2 文件共享实现

from multiprocessing import Process, Lock
import timedef sell_tickets(lock, window):while True:with lock:with open("tickets.txt", "r+") as f:tickets = int(f.read())if tickets <= 0:breaktime.sleep(0.1)print(f"{window}售出票号{tickets}")f.seek(0)f.write(str(tickets-1))if __name__ == '__main__':lock = Lock()# 初始化票数with open("tickets.txt", "w") as f:f.write("100")processes = [Process(target=sell_tickets, args=(lock, f"窗口{i}"))for i in range(1, 4)]for p in processes:p.start()for p in processes:p.join()print("所有票已售完")

6. 同步机制选择指南

机制适用场景优点缺点
Lock互斥访问共享资源简单直接可能引起死锁
Semaphore限制资源访问数量灵活控制并发度比锁复杂
Event进程间简单通知轻量级通知机制不传递数据

7. 常见问题解答

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

  • 锁是二元信号量(0/1)，信号量可以设置多个许可

  • 锁用于互斥访问，信号量用于控制资源访问数量

Q2: 如何避免死锁？

  • 按固定顺序获取多个锁

  • 使用超时机制

  • 避免嵌套锁

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

  • 共享内存更快但更复杂

  • 文件共享更简单但性能较低

  • 根据场景选择合适方案

相关文章：

  • 第十六节：高频开放题-React与Vue设计哲学差异
  • 闭包函数的应用与理解
  • iOS 应用性能测试工具对比：Xcode Instruments、克魔助手与性能狗
  • FileZilla“服务器发回了不可路由的地址，使用服务器地址代替
  • 专题讨论：BST树上的添加与删除
  • Few-shot medical image segmentation with high-fidelity prototypes 论文总结
  • 使用tshark命令解析tcpdump抓取的数据包
  • 【计量地理学】实验六 地理属性空间插值
  • OpenCV 模板与多个对象匹配方法详解（继OpenCV 模板匹配方法详解）
  • 【PyTorch】训练时跟OOM相关的提示信息
  • 传导发射测试(CE)和传导骚扰抗扰度测试(CS)
  • Unity3D仿星露谷物语开发36之锄地动画2
  • 【C++】类和对象之日期类的实现（构造、运算符重载）
  • 机器学习中的“三态模型“：过拟合、欠拟合和刚刚好
  • 在FVM（有限体积法）的CFD仿真中，AI和机器学习的应用
  • 关于进程状态
  • 计算机组成原理笔记(十七)——4.2定点加减运算
  • docker配置skywalking 监控springcloud应用
  • Laravel-vite+vue开发前端模板
  • MIT6.S081-lab4
  • 马上评｜治理“龟速车”，也是一次驾驶文明的升级
  • 北京理工大学：教师宫某涉嫌师德失范，暂停其一切职务活动
  • 日本长野一夜连震47次，当局呼吁警惕更大地震
  • 观察｜雀巢咖啡加码中国布局，如何借势云南咖啡打造新增长极？
  • 二手服装“批发”市集受到年轻人追捧，是哪一股潮流在推动？
  • 奥利弗·沙赫特博士：集群是产业集聚地，更是“超级连接器”