37.Python之生产者消费者模型

Python之生产者消费者模型

1. 生产者消费者模型

   • 模型指的是一种解决问题的套路，目的是为了使生产数据与处理数据达到平衡，使得效率最大化。

   ---

2. 生产者消费者模型中包含两类重要角色一类叫生产者，另一类叫消费者：

   • 生产者：将负责制造数据的称为生产者（生产数据）。通常在生产数据之前需要通过一些代码濑获取数据。
   • 消费者：接收生产者制造出的数据，来做进一步处理，该类任务被比喻成消费者（处理数据）。通常在获取代码之后需要通过一些代码进行数据处理。

   ---

3. 实现生产者消费者模型的三要素：

   • 生产者
   • 消费者
   • 队列（队列中存放的是一些消息）

   ---

4. 生产者消费者模型的运作方式：

   • 生产者生产数据，放到一个共享的空间，然后消费者取走进行处理。

   ---

5. 生产者消费者模型的实现方式：（由于生产者消费者模型并不局限于某一类技术，因此，有多种实现方式，不限于以下方式）

   • 生产者进程+队列+消费者进程

   ---

6. 该模型的应用场景

   • 程序中出现明显的两类任务，一类任务负责生产数据，另一类任务是负责处理生产数据的，此时就应该考虑生产者消费者模型。（例如：爬虫）

   ---

7. 使用生产者消费者模型的优势：

   • 实现了生产者与消费者解耦合
   • 平衡了生产力和消费力，彼此不影响，生产者可以一致不停的生产，消费者可以一致不停的消费，因为二者不再是直接沟通了，而是跟队列沟通

   import time
   import random
   from multiprocessing import Process,Queue
   
   def consumer(name, q):
    while True:
        res = q.get()
        if res == None:break
        time.sleep(random.randint(1,3))
        print(F'[{name}]获取到{res}')
   
   def producer(name,q,tools):
    for i in range(3):
        time.sleep(random.randint(1,2))
        res = F'{tools}:{i}'
        q.put(res)
        print(F'[{name}]制造了{res}')
    q.put(None)
   
   if __name__ == '__main__':
    # 队列
    q = Queue()
   
    # 生产者
    p1 = Process(target=producer, args=(('生产者-1', q, '食物')))
    p2 = Process(target=producer, args=(('生产者-2', q, '武器')))
   
   
    # 消费者
    c1 = Process(target=consumer, args=(('消费者-1',q)))
    c2 = Process(target=consumer, args=(('消费者-2',q)))
    c3 = Process(target=consumer, args=(('消费者-3',q)))
   
   
    p1.start()
    p2.start()
   
    c1.start()
    c2.start()
    c3.start()
   
    # 在生产者生产完毕之后，往队列的末尾添加一个结束信号None
    p1.join()
    p2.join()
   
    # 有几个消费者就应该放几个结束信号
    q.put(None)
    q.put(None)
    q.put(None)
   

   ---

   改进：（加入守护进程）

   import time
   import random
   from multiprocessing import Process,JoinableQueue
   
   def consumer(name, q):
      while True:
         res = q.get()
         if res == None:break
         time.sleep(random.randint(1,3))
         print(F'[{name}]获取到{res}')
         q.task_done()
   
   def producer(name, q, tools):
      for i in range(3):
         time.sleep(random.randint(1,2))
         res = F'{tools}:{i}'
         q.put(res)
         print(F'[{name}]制造了{res}')
   
   if __name__ == '__main__':
      # 队列
      q = JoinableQueue()
   
      # 生产者
      p1 = Process(target=producer, args=(('生产者-1', q, '食物')))
      p2 = Process(target=producer, args=(('生产者-2', q, '武器')))
   
   
      # 消费者
      c1 = Process(target=consumer, args=(('消费者-1', q)))
      c2 = Process(target=consumer, args=(('消费者-2', q)))
      c3 = Process(target=consumer, args=(('消费者-3', q)))
      c1.daemon = True
      c2.daemon = True
      c3.daemon = True
   
   
      p1.start()
      p2.start()
   
      c1.start()
      c2.start()
      c3.start()
      # 确定生产者生产完毕
      p1.join()
      p2.join()
      # 在生产者生产完毕后，拿到队列中元素的总个数，然后知道元素总数变为0，q.join()这一行代码才算运行完毕
      q.join()
      print('父进程结束')