一 生产者消费者模型介绍
为什么要使用生产者消费者模型
生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者和消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的
二 生产者消费者模型实现
基于上一小节学习的队列来实习一个生产者消费者模型
from multiprocessing import Process,Queue import time def product(q): for i in range(10): time.sleep(0.5) print("生产了 %s ge 包子" % i) q.put(i) def consumer(q): while True: time.sleep(1) res = q.get() if res is None: print("包子都吃完了") break print("消费了--> %s ge 包子" % res) if __name__ == "__main__": q = Queue() p1 = Process(target=product,args=(q,)) c1 = Process(target=consumer,args=(q,)) p1.start() c1.start() p1.join() q.put(None) # 在生产者生产所有东西最后在队列最后添加一个None，要不会阻塞，
# 此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环
结果是
生产了 0 ge 包子 生产了 1 ge 包子 消费了--> 0 ge 包子 生产了 2 ge 包子 生产了 3 ge 包子 消费了--> 1 ge 包子 生产了 4 ge 包子 生产了 5 ge 包子 消费了--> 2 ge 包子 生产了 6 ge 包子 生产了 7 ge 包子 消费了--> 3 ge 包子 生产了 8 ge 包子 生产了 9 ge 包子 消费了--> 4 ge 包子 消费了--> 5 ge 包子 消费了--> 6 ge 包子 消费了--> 7 ge 包子 消费了--> 8 ge 包子 消费了--> 9 ge 包子 包子都吃完了
多个生产者，消费者
from multiprocessing import Process,Queue import time def product(q): for i in range(10): time.sleep(0.5) print("生产了 %s 个 包子" % i) q.put(i) def consumer(q): while True: time.sleep(1) res = q.get() if res is None: print("包子都吃完了") break print("消费了--> %s 个 包子" % res) if __name__ == "__main__": q = Queue() p1 = Process(target=product,args=(q,)) p2 = Process(target=product,args=(q,)) p3 = Process(target=product,args=(q,)) c1 = Process(target=consumer,args=(q,)) c2 = Process(target=consumer,args=(q,)) p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.put(None) q.put(None) #最后是2个None，几个消费者就几个None
结果
C:Python35python.exe D:/luffy/第四模块/10生产者消费者模型.py 生产了 0 个 包子 生产了 0 个 包子 生产了 0 个 包子 生产了 1 个 包子 生产了 1 个 包子 生产了 1 个 包子 消费了--> 0 个 包子 消费了--> 0 个 包子 生产了 2 个 包子 生产了 2 个 包子 生产了 2 个 包子 生产了 3 个 包子 生产了 3 个 包子 生产了 3 个 包子 消费了--> 0 个 包子 消费了--> 1 个 包子 生产了 4 个 包子 生产了 4 个 包子 生产了 4 个 包子 生产了 5 个 包子 生产了 5 个 包子 生产了 5 个 包子 消费了--> 1 个 包子 消费了--> 1 个 包子 生产了 6 个 包子 生产了 6 个 包子 生产了 6 个 包子 生产了 7 个 包子 生产了 7 个 包子 生产了 7 个 包子 消费了--> 2 个 包子 消费了--> 2 个 包子 生产了 8 个 包子 生产了 8 个 包子 生产了 8 个 包子 生产了 9 个 包子 生产了 9 个 包子 生产了 9 个 包子 消费了--> 2 个 包子 消费了--> 3 个 包子 消费了--> 3 个 包子 消费了--> 3 个 包子 消费了--> 4 个 包子 消费了--> 4 个 包子 消费了--> 4 个 包子 消费了--> 5 个 包子 消费了--> 5 个 包子 消费了--> 5 个 包子 消费了--> 6 个 包子 消费了--> 6 个 包子 消费了--> 6 个 包子 消费了--> 7 个 包子 消费了--> 7 个 包子 消费了--> 7 个 包子 消费了--> 8 个 包子 消费了--> 8 个 包子 消费了--> 8 个 包子 消费了--> 9 个 包子 消费了--> 9 个 包子 消费了--> 9 个 包子 包子都吃完了 包子都吃完了
但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决,有几个消费者就需要发送几次结束信号：相当low,例如
JoinableQueue([maxsize])
这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
参数介绍
maxsize是队列中允许最大项数，省略则无大小限制。
方法介绍
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止
基于JoinableQueue实现生产者消费者模型
from multiprocessing import Process,JoinableQueue import time def product(q): for i in range(2): time.sleep(0.5) print("生产了 %s 个 包子" % i) q.put(i) q.join() def consumer(q): while True: res = q.get() if res is None: print("包子都吃完了") break time.sleep(1) print("消费了--> %s 个 包子" % res) q.task_done() if __name__ == "__main__": q = JoinableQueue() p1 = Process(target=product,args=(q,)) p2 = Process(target=product,args=(q,)) p3 = Process(target=product,args=(q,)) c1 = Process(target=consumer,args=(q,)) c2 = Process(target=consumer,args=(q,)) c1.daemon = True c2.daemon = True p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join()
运行结果是
生产了 0 个 包子 生产了 0 个 包子 生产了 0 个 包子 生产了 1 个 包子 生产了 1 个 包子 生产了 1 个 包子 消费了--> 0 个 包子 消费了--> 0 个 包子 消费了--> 0 个 包子 消费了--> 1 个 包子 消费了--> 1 个 包子 消费了--> 1 个 包子
生产者消费者模型总结
1、程序中有两类角色
一类负责生产数据（生产者）
一类负责处理数据（消费者）
2、引入生产者消费者模型为了解决的问题是
平衡生产者与消费者之间的速度差
程序解开耦合
3、如何实现生产者消费者模型
生产者<--->队列<--->消费者