Python-----进程通信队列

1. 互斥锁

   from multiprocessing import Process
   from multiprocessing import Lock
   import time
   import random
   def task1(lock):
       print('task1')
       lock.acquire()
       print('task1:开始打印')
       time.sleep(random.randint(1,3))
       print('task1:打印完成')
       lock.release()
   def task2(lock):
       print('task2')
       lock.acquire()
       print('task2:开始打印')
       time.sleep(random.randint(1,3))
       print('task2:打印完成')
       lock.release()
   def task3(lock):
       print('task3')
       lock.acquire()
       print('task3:开始打印')
       time.sleep(random.randint(1,3))
       print('task3:打印完成')
       lock.release()
   if __name__ == '__main__':
       lock = Lock()
       p1 = Process(target=task1,args=(lock,))
       p2 = Process(target=task2, args=(lock,))
       p3 = Process(target=task3, args=(lock,))
       p1.start()
       p2.start()
       p3.start()
   

2. 进程之间的通信: 队列.

   上一节: 多个进程之间的通信:基于文件以及加锁的方式.

   1. 操作文件效率低.
   2. 自己加锁很麻烦. 很容易出现死锁,递归锁.

   进程之间的通信最好的方式是基于队列.

   什么是队列?

   队列就是存在于内存中的一个容器,最大的一个特点; 队列的特性就是FIFO.完全支持先进先出的原则.

   from multiprocessing import Queue
   # q = Queue(3)  # 可以设置元素个数
   # def func():
   #     print('in func')
   # q.put('alex')
   # q.put({'count': 1})
   # q.put(func)
   # print(222)
   # q.put(666)  # 当队列数据已经达到上限,在插入数据的时候,程序就会夯住.
   # print(111)
   # print(q.get())
   # print(q.get())
   # print(q.get())
   # ret = q.get()
   # ret()
   # 当你将数据取完继续在取值时,默认也会阻塞
   # print(q.get())
   # print(q.get())
   # print(q.get())
   # print(q.get())
   

   其他参数:

   1. maxsize() q = Queue(3) 数据量不易过大.精简的重要的数据.

   2. put block 默认为True 当你插入的数据超过最大限度,默认阻塞

      # q = Queue(3)  # 可以设置元素个数
      # q.put('alex')
      # q.put({'count': 1})
      # q.put(22)  #
      # q.put(333,block=False)  # 改成False 数据超过最大限度,不阻塞了直接报错.
      

   3. put timeout() 参数 :延时报错,超过三秒再put不进数据,就会报错.

      get里面的参数:block,timeout一样.

      # q = Queue(3)  # 可以设置元素个数
      # q.put('alex')
      # q.put({'count': 1})
      # q.put(22)  #
      # q.put(333,timeout=3)  # 延时报错,超过三秒再put不进数据,就会报错.
      # q = Queue()
      # q.put('alex')
      # q.put({'count': 1})
      # q.put(22)  #
      # print(q.get())
      # print(q.get())
      # print(q.get())
      # print(q.get(block=False))  timeout
      

3. 进程之间的通信实例.

   # 小米:抢手环4.预期发售10个.
   # 有100个人去抢.
   import os
   from multiprocessing import Queue
   from multiprocessing import Process
   def task(q):
       try:
           q.put(f'{os.getpid()}',block=False)
       except Exception:
           return
   if __name__ == '__main__':
       q = Queue(10)
       for i in range(100):
           p = Process(target=task,args=(q,))
           p.start()
       for i in range(1,11):
           print(f'排名第{i}的用户: {q.get()}',)
   

   利用队列进行进程之间通信: 简单,方便,不用自己手动加锁.队列自带阻塞,可持续化取数据.

4. 生产者消费者模型

   模型, 设计模式,归一化设计, 理论等等,教给你一个编程思路.如果以后遇到类似的情况,直接套用即可.

   生产者: 生产数据进程.

   消费者: 对生产者生产出来的数据做进一步处理进程.

   吃饭: 吃包子. 厨师生产出包子,不可能直接给到你嘴里. 放在一个盆中,消费者从盆中取出包子食用.三个主体 : (生产者)厨师, (容器队列)盆, (消费者)人.

   为什么夹杂这个容器?

   如果没有容器, 生产者与消费者强耦合性.不合理.所以我们要有一个容器,缓冲区.平衡了生产力与消费力.

   生产者消费者多应用于并发.

   from multiprocessing import Process
   from multiprocessing import Queue
   import time
   import random
   def producer(name,q):
       for i in range(1,6):
           time.sleep(random.randint(1,3))
           res = f'{i}号包子'
           q.put(res)
   
           print(f'33[0;32m 生产者{name}: 生产了{res}33[0m')
   
   def consumer(name,q):
       while 1:
           try:
               time.sleep(random.randint(1,3))
               ret = q.get(timeout=5)
               print(f'消费者{name}: 吃了{ret}')
           except Exception:
               return
   if __name__ == '__main__':
       q = Queue()
       p1 = Process(target=producer, args=('太白',q))
       p2 = Process(target=consumer, args=('MC骚强',q))
       p1.start()
       p2.start()
   

   生产者消费者模型:

   合理的去调控多个进程去生成数据以及提取数据,中间有个必不可少的环节容器队列.