day 7-5 生产者消费者模型

一. 生产者和消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

二. 为什么要使用生产者和消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

三. 什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

四. 基于队列实现生产者消费者模型

from multiprocessing import Queue,Process
import time

def pro(p):
    for  i in range(3):
        res = "包子%s" %i
        time.sleep(2)
        print("生产者生产了: %s" %res)
        p.put(res)

def con(p):
    while True:
        res = p.get()
        time.sleep(1)
        print("消费者吃了%s"%res)


if __name__ == '__main__':
    #容器,一个队列
    p =Queue()

    #生产者们(可以有多个生产者)
    p1 = Process(target=pro,args=(p,))


    #消费者们,也可以有多个
    c1 = Process(target=con,args=(p,))

    # 启动
    p1.start()
    c1.start()
    print("主进程...")

　　此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

　　

from multiprocessing import Queue,Process
 import time
 
 def pro(p):
        for  i in range(3):
          res = "包子%s" %i
          time.sleep(2)
          print("生产者生产了: %s" %res)
          p.put(res)
 
 def con(p):
     while True:
         res = p.get()
         if res == None:    #在这里判断,如果最后的数据是None就退出
             break
         time.sleep(1)
         print("消费者吃了%s"%res)
 
 
 if __name__ == '__main__':
     #容器,一个队列
     p =Queue()
 
     #生产者们(可以有多个生产者)
     p1 = Process(target=pro,args=(p,))
 
 
     #消费者们,也可以有多个
     c1 = Process(target=con,args=(p,))

     # 启动
     p1.start()
     c1.start()
     p1.join()
     p.put(None)    #在确保生产者运行结束后,我们往队列里放入一个None,然在消费者里做一个判断
     print("主进程...")

 Joinablequeue:

　　JoinableQueue([maxsize])

这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
参数介绍:maxsize是队列中允许最大项数，省略则无大小限制。

方法介绍

　　JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
　　q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
　　q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

　　基于JoinableQueue实现生产者消费者模型

from multiprocessing import JoinableQueue,Process
import time

def pro(p):
    for  i in range(3):
        res = "包子%s" %i
        time.sleep(2)
        print("生产者生产了: %s" %res)
        p.put(res)
    p.join()            # 队列取空了.进程才算结束.那么c1进程也就没有存在的必要了.
def con(p):
    while True:
        res = p.get()
        if res == None:
            break
        time.sleep(1)
        print("消费者吃了%s"%res)
        p.task_done()


if __name__ == '__main__':
    #容器,一个队列
    p =JoinableQueue()

    #生产者们(可以有多个生产者)
    p1 = Process(target=pro,args=(p,))


    #消费者们,也可以有多个
    c1 = Process(target=con,args=(p,))
    c1.daemon =True     #c1设置成守护进程
    # 启动
    p1.start()
    c1.start()
    p1.join()

    print("主进程...")

 p.join()本质是等待进程p执行完毕.在pro中使用q.join()表示等待队列执行完毕.队列只有全部取完的时候,才算执行完毕.