Python下进程同步之互斥锁信号量事件机制 𪕽

Posted bbb001

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python下进程同步之互斥锁信号量事件机制 𪕽相关的知识，希望对你有一定的参考价值。

原文: http://blog.gqylpy.com/gqy/229

"


一、锁机制:??multiprocess.Lock

上篇博客中，我们千方百计实现了程序的异步，让多个任务同时在几个进程中并发处理，但它们之间的运行没有顺序。尽管并发编程让我们能更加充分的利用io资源，但是也给我我们带来了新问题，多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题. 例：

  1. # 多进程抢占输出资源
  2.  
  3. from multiprocessing import Process
  4. from os import getpid
  5. from time import sleep
  6. from random import random
  7.  
  8.  
  9. def work(i):
  10.     print("%s: %s is running" %(i, getpid()))
  11.     sleep(random())
  12.     print("%s: %s is done" %(i, getpid()))
  13.  
  14.  
  15. if __name__ == ‘__main__‘:
  16.     for i in range(5):
  17.         p = Process(target=work, args=(i,))
  18.         p.start()

使用互斥锁维护执行顺序：

  1. # 使用锁机制维护执行顺序
  2.  
  3. from multiprocessing import Process, Lock
  4. from os import getpid
  5. from time import sleep
  6. from random import random
  7.  
  8.  
  9. def work(lock, i):
  10.     lock.acquire()    # ??开锁进门
  11.     print("%s: %s is running" %(i, getpid()))
  12.     sleep(random())
  13.     print("%s: %s is done" %(i, getpid()))
  14.     lock.release()    # ??出门，归还钥匙
  15.  
  16. if __name__ == ‘__main__‘:
  17.     lock = Lock()    # ??实例化一把锁，一把钥匙
  18.  
  19.     for i in range(5):
  20.         p = Process(target=work, args=(lock, i))
  21.         p.start()

上面这种情况虽然使用加锁的形式实现了顺序的执行，却使得进程变成了串行执行，这样确实会浪费些时间，但是保证了数据的安全.

接下来我们以模拟抢票为例，来看看数据安全的重要性：

  1. # 文件db的内容为："count": 1 （注意："count"一定要用双引号，否者json无法识别）
  2. # 并发运行，效率高，但是：竞争抢票只读写同一个文件，造成数据写入错乱
  3.  
  4. from multiprocessing import Process
  5. from time import sleep
  6. from json import load, dump
  7.  
  8.  
  9. def search():
  10.     dct = load(open(‘db‘))
  11.     print("\033[43m剩余票数:%s\033[0m" %dct[‘count‘])
  12.  
  13.  
  14. def get():
  15.     dct = load(open(‘db‘))
  16.     sleep(0.01)  # 模拟读数据的网络延迟
  17.  
  18.     if dct[‘count‘] > 0:
  19.         dct[‘count‘] -=1
  20.         sleep(0.02)  # 模拟写数据的网络延迟
  21.         dump(dct, open(‘db‘, ‘w‘))
  22.         print("\033[32m购票成功\033[0m")
  23.  
  24.  
  25. def tack():
  26.     search()
  27.     get()
  28.  
  29.  
  30. if __name__ == ‘__main__‘:
  31.     for i in range(100):    # 模拟并发100个客户端抢票
  32.         p =Process(target=tack)
  33.         p.start()
  34.  
  35. # 100个进程同时读写同一个文件，可能会报错

使用锁机制保护数据安全：

  1. # 文件db的内容为："count": 1 （注意："count"一定要用双引号，否者json无法识别）
  2. # 同步运行，效率低，但是可以保证数据安全
  3.  
  4. from multiprocessing import Process, Lock
  5. from time import sleep
  6. from json import load, dump
  7.  
  8.  
  9. def search():
  10.     dct = load(open(‘db‘))
  11.     print("\033[43m剩余票数:%s\033[0m" %dct[‘count‘])
  12.  
  13.  
  14. def get():
  15.     dct = load(open(‘db‘))
  16.     sleep(0.01)  # 模拟读数据的网络延迟
  17.  
  18.     if dct[‘count‘] > 0:
  19.         dct[‘count‘] -=1
  20.         sleep(0.02)  # 模拟写数据的网络延迟
  21.         dump(dct, open(‘db‘, ‘w‘))
  22.         print("\033[32m购票成功\033[0m")
  23.  
  24.  
  25. def tack():
  26.     search()
  27.     lock.acquire()    # 上锁
  28.     get()
  29.     lock.release()    # 释放
  30.  
  31.  
  32. if __name__ == ‘__main__‘:
  33.     lock = Lock()    # 实例化一把锁
  34.  
  35.     for i in range(100):    # 模拟并发100个客户端抢票
  36.         p =Process(target=tack)
  37.         p.start()
  38.  
  39. # 可能会遇到json.decoder.JSONDecodeError报错

使用锁机制可以保证多个进程想要修改同一块数据时，在同一时间点只能有一个进程进行修改，即串行的修改，牺牲速度而保证安全性.

虽然可以用文件共享数据实现进程间通讯，但问题是：1. 效率低（共享数据基于文件，而文件是硬盘上的数据）；2. 需要自己加锁处理。因此我们最好找寻一种解决方案能够兼顾：1. 效率高（多个进程共享一块内存的数据）；2. 帮我们处理好锁的问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道

队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的问题中解脱出来，我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可扩展性.

二、信号量：multiprocessing.Semaphore

1. 信号量Semaphore允许一定数量的线程在同一时间点更改同一块数据，很形象的例子：厕所里有3个坑，同时可以3个人蹲，如果来了第4个人就要在外面等待，直到某个人出来了才能进去.

2.信号量同步基于内部计数器，每调用一次acquire()，计数器-1，每调用一次release()，计数器+1，当计数器为0时，acquire()调用被阻塞。这是迪克斯彻（Dijkstra）信号量概念P()和V()的Python实现，信号同步机制适用于访问像服务器这样的有限资源.

3.注意：信号量与进程池的概念很像，要注意区分，信号量涉及到加锁的概念.

  • 方法

obj = Semaphore(4)：实例化一把锁，4把钥匙，钥匙可以指定（最小0，最大未知）

obj.acquire()：加锁，锁住下方的语句，直到遇到obj.release()方可解锁

obj.release()：释放，释放obj.acquire()和obj.release()的语句数据，此方法如果写在加锁之前，便多了一把钥匙

  • 基本用法
  1. # 一把锁，多把钥匙
  2.  
  3. from multiprocessing import Semaphore
  4.  
  5. s = Semaphore(2)    # 实例化一把锁，配2把钥匙
  6.  
  7. s.release()     # 可以先释放钥匙，变成3把钥匙
  8. s.release()     # 再释放一把钥匙，现在变成了4把钥匙
  9.  
  10. s.acquire()
  11. print(1)
  12.  
  13. s.acquire()
  14. print(2)
  15.  
  16. # 加上先释放的钥匙，我门总共有4把钥匙
  17. s.acquire() # 顺利执行
  18. print(3)
  19.  
  20. s.acquire() # 顺利执行
  21. print(4)
  22.  
  23. # 钥匙全部被占用
  24. s.acquire() # 此处阻塞住，等待钥匙的释放
  25. print(5)    # 不会被打印
  • 进阶：小黑屋
  1. # 小黑屋
  2.  
  3. from multiprocessing import Process, Semaphore
  4. from time import sleep
  5. from random import uniform
  6.  
  7.  
  8. def func(sem, i):
  9.     sem.acquire()
  10.     print("第%s个人进入了小黑屋" % i)
  11.     sleep(uniform(1, 3))
  12.     print("第%s个人走出了小黑屋" % i)
  13.     sem.release()
  14.  
  15.  
  16. if __name__ == ‘__main__‘:
  17.     sem = Semaphore(5)  # 初始化一把锁，配5把钥匙
  18.  
  19.     for i in range(10): # 启动10个子进程，最多只能5个人同在小黑屋中
  20.         p = Process(target=func, args=(sem, i))
  21.         p.start()

三、事件机制：multiprocessing.Event

Python线程的事件用于主线程控制其它线程的执行，事件主要提供了三个方法：set、wait，clear

事件处理机制：全局定义了一个“Flag”，如果“Flag”值为False，程序执行wait()方法会被阻塞；如果“Flag”值为True，程序执行wait()方法便不会被阻塞.

  • 方法

obj.is_set()：默认值为False，事件是通过此方法的bool值去标示wait()的阻塞状态

obj.set()：将is_set()的bool值改为True

obj.clear()：将is_set()的bool值改为False

obj.wait()：is_set()的值为False时阻塞，否则不阻塞

  • 实例：模拟红绿灯
  1. # 模拟红绿灯
  2.  
  3. from multiprocessing import Process, Event
  4. import time
  5. import random
  6.  
  7.  
  8. def Tra(e):
  9.     print("\033[32m绿灯亮\033[0m")
  10.     e.set()
  11.  
  12.     while 1:
  13.         if e.is_set():
  14.             time.sleep(3)
  15.             print("\033[31m红灯亮\033[0m")
  16.             e.clear()
  17.         else:
  18.             time.sleep(3)
  19.             print("\033[32m绿灯亮\033[0m")
  20.             e.set()
  21.  
  22.  
  23. def Car(e, i):
  24.     e.wait()
  25.     print("第%s辆小汽车过去了" % i)
  26.  
  27.  
  28. if __name__ == ‘__main__‘:
  29.     e = Event()
  30.  
  31.     tra = Process(target=Tra, args=(e,))
  32.     tra.start()
  33.  
  34.     for i in range(100):    # 模拟一百辆小汽车
  35.         time.sleep(0.5)
  36.         car = Process(target=Car, args=(e, i))
  37.         car.start()
        </div>

"

原文: http://blog.gqylpy.com/gqy/229

以上是关于Python下进程同步之互斥锁信号量事件机制 𪕽的主要内容，如果未能解决你的问题，请参考以下文章

Python下进程同步之互斥锁信号量事件机制 -- 2019-08-16 18:05:40

Python之路(第三十八篇) 并发编程：进程同步锁/互斥锁信号量事件队列生产者消费者模型

Python多进程,同步互斥,信号量,锁补充上一篇文章

Python之进程同步控制（锁信号量事件 ）进程间通信——队列和管道

同步和互斥

同步互斥按键驱动