进程之生产者消费者模型

1、什么是生产者消费者模型

　　生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，若是生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。一样的道理，若是消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题因而引入了生产者和消费者模式。

　　该模型的工做方式：生产者生产数据传递给消费者处理

　　实现方式：生产者 ---> 队列 <--- 消费者

2、为什么要使用生产者消费者模型

　　当程序中出现明显的两类任务，一类负责生产数据，一类负责处理数据，就能够引入生产者消费者模型来实现生产者与消费者的解耦合，平衡生产能力与消费能力，从而提高效率

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

　　这个阻塞队列就是用来给生产者和消费者解耦的

3、生产者消费者模型的使用

基于队列来实现一个生产者消费者模型

import time, random from multiprocessing import Process, Queue def producer(name, food, q): for i in range(6): res = "%s%s" %(food, i) time.sleep(random.randint(1, 3)) # 模拟生产数据的时间
 q.put(res) print("厨师 %s 生产了 %s" %(name, res)) def consumer(name, q): while True: res = q.get() time.sleep(random.randint(1, 3)) # 模拟处理数据的时间
        print("吃货 %s 吃了 %s" % (name, res)) if __name__ == '__main__': q = Queue() # 生产者
    p1 = Process(target=producer, args=("qiu", "包子", q)) # 消费者
    c1 = Process(target=consumer, args=("xi", q)) p1.start() c1.start() print("主进程") # 运行结果
主进程 厨师 qiu 生产了 包子0 吃货 xi 吃了 包子0 厨师 qiu 生产了 包子1 厨师 qiu 生产了 包子2 吃货 xi 吃了 包子1 吃货 xi 吃了 包子2 厨师 qiu 生产了 包子3 吃货 xi 吃了 包子3 厨师 qiu 生产了 包子4 吃货 xi 吃了 包子4 厨师 qiu 生产了 包子5 吃货 xi 吃了 包子5

View Code

此时的问题是主进程永远不会结束，缘由是：生产者 p 在生产完后就结束了，可是消费者 c 在取空了 q 以后，则一直处于死循环中且卡在 q.get() 这一步

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就能够 break 出死循环

import time, random from multiprocessing import Process, Queue def producer(name, food, q): for i in range(3): res = "%s%s" %(food, i) time.sleep(random.randint(1, 3)) # 模拟生产数据的时间
 q.put(res) print("厨师 %s 生产了 %s" %(name, res)) q.put(None) def consumer(name, q): while True: res = q.get() if not res: break time.sleep(random.randint(1, 3)) # 模拟处理数据的时间
        print("吃货 %s 吃了 %s" % (name, res)) if __name__ == '__main__': q = Queue() # 生产者
    p1 = Process(target=producer, args=("qiu", "包子", q)) # 消费者
    c1 = Process(target=consumer, args=("xi", q)) p1.start() c1.start() print("主进程") # 运行结果
主进程 厨师 qiu 生产了 包子0 吃货 xi 吃了 包子0 厨师 qiu 生产了 包子1 吃货 xi 吃了 包子1 厨师 qiu 生产了 包子2 吃货 xi 吃了 包子2

View Code

上述解决方式，在有多个生产者和多个消费者时，会出现生产完了但没有消费的状况

import time, random from multiprocessing import Process, Queue def producer(name, food, q): for i in range(3): res = "%s%s" %(food, i) time.sleep(random.randint(1, 3)) # 模拟生产数据的时间
 q.put(res) print("厨师 %s 生产了 %s" %(name, res)) q.put(None) def consumer(name, q): while True: res = q.get() if not res: break time.sleep(random.randint(1, 3)) # 模拟处理数据的时间
        print("吃货 %s 吃了 %s" % (name, res)) if __name__ == '__main__': q = Queue() # 生产者
    p1 = Process(target=producer, args=("qiu", "包子", q)) p2 = Process(target=producer, args=("xi", "饺子", q)) p3 = Process(target=producer, args=("ru", "馒头", q)) # 消费者
    c1 = Process(target=consumer, args=("feng", q)) c2 = Process(target=consumer, args=("yan", q)) p1.start() p2.start() p3.start() c1.start() c2.start() print("主进程") # 运行结果
主进程 厨师 qiu 生产了 包子0 厨师 qiu 生产了 包子1 厨师 ru 生产了 馒头0 厨师 qiu 生产了 包子2 厨师 xi 生产了 饺子0 厨师 ru 生产了 馒头1 吃货 feng 吃了 包子0 厨师 ru 生产了 馒头2 吃货 yan 吃了 包子1 吃货 yan 吃了 包子2 厨师 xi 生产了 饺子1 吃货 feng 吃了 馒头0 厨师 xi 生产了 饺子2 吃货 feng 吃了 饺子0 吃货 feng 吃了 馒头1 吃货 feng 吃了 馒头2

View Code

可见，生产了饺子1 和饺子 2，可是没有人吃。其实这是由于生产者 1 生产完就往队列里放，生产者 2 生产完也往队列里放，3 也是，这是同一个队列，但三个生产者的 for 循环并非同时结束，因此要把结束信号放到数据的最后

import time, random from multiprocessing import Process, Queue def producer(name, food, q): for i in range(3): res = "%s%s" %(food, i) time.sleep(random.randint(1, 3)) # 模拟生产数据的时间
 q.put(res) print("厨师 %s 生产了 %s" %(name, res)) def consumer(name, q): while True: res = q.get() if not res: break time.sleep(random.randint(1, 3)) # 模拟处理数据的时间
        print("吃货 %s 吃了 %s" % (name, res)) if __name__ == '__main__': q = Queue() # 生产者
    p1 = Process(target=producer, args=("qiu", "包子", q)) p2 = Process(target=producer, args=("xi", "饺子", q)) p3 = Process(target=producer, args=("ru", "馒头", q)) # 消费者
    c1 = Process(target=consumer, args=("feng", q)) c2 = Process(target=consumer, args=("yan", q)) p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.put(None) q.put(None) print("主进程") # 运行结果
厨师 qiu 生产了 包子0 厨师 ru 生产了 馒头0 厨师 xi 生产了 饺子0 吃货 feng 吃了 包子0 厨师 qiu 生产了 包子1 厨师 xi 生产了 饺子1 吃货 yan 吃了 馒头0 厨师 ru 生产了 馒头1 吃货 feng 吃了 饺子0 吃货 yan 吃了 包子1 吃货 feng 吃了 饺子1 厨师 qiu 生产了 包子2 厨师 xi 生产了 饺子2 吃货 yan 吃了 馒头1 厨师 ru 生产了 馒头2 主进程 吃货 yan 吃了 饺子2 吃货 feng 吃了 包子2 吃货 yan 吃了 馒头2

View Code

但这种方式很是 low，有几个消费者就须要发送几回结束信号，因此有另一种队列提供了解决方案

JoinableQueue([maxsize])

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

方法介绍：JoinableQueue 的实例 q 除了与 Queue 对象相同的方法以外还具备：

　　q.task_done()：使用者使用此方法发出信号，表示 q.get() 的返回项目已经被处理。若是调用此方法的次数大于从队列中删除项目的数量，将引起 ValueError 异常

　　q.join()：生产者调用此方法进行阻塞，直到队列中全部的项目均被处理。阻塞将持续到队列中的每一个项目均调用 q.task_done() 方法为止

import time, random from multiprocessing import Process, JoinableQueue def producer(name, food, q): for i in range(3): res = "%s%s" %(food, i) time.sleep(random.randint(1, 3)) # 模拟生产数据的时间
 q.put(res) print("厨师 %s 生产了 %s" %(name, res)) def consumer(name, q): while True: res = q.get() time.sleep(random.randint(1, 3)) # 模拟处理数据的时间
        print("吃货 %s 吃了 %s" % (name, res)) q.task_done() if __name__ == '__main__': q = JoinableQueue() # 生产者
    p1 = Process(target=producer, args=("qiu", "包子", q)) p2 = Process(target=producer, args=("xi", "饺子", q)) p3 = Process(target=producer, args=("ru", "馒头", q)) # 消费者
    c1 = Process(target=consumer, args=("feng", q)) c2 = Process(target=consumer, args=("yan", q)) # 主进程等生产者p一、p二、p3结束, 而p一、p二、p3是在消费者把全部数据都取干净以后才会结束
    # 因此一旦p一、p二、p3结束了，证实消费者也不必存在了，应该随着主进程一块死掉，于是须要将消费者设置成守护进程
    c1.daemon = True c2.daemon = True p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.join() # 主进程等q结束, 即q内的数据被取干净了
    print("主进程") # 运行结果
厨师 xi 生产了 饺子0 厨师 ru 生产了 馒头0 厨师 qiu 生产了 包子0 厨师 xi 生产了 饺子1 吃货 feng 吃了 饺子0 厨师 qiu 生产了 包子1 厨师 qiu 生产了 包子2 厨师 xi 生产了 饺子2 吃货 yan 吃了 馒头0 厨师 ru 生产了 馒头1 吃货 feng 吃了 包子0 吃货 yan 吃了 饺子1 厨师 ru 生产了 馒头2 吃货 feng 吃了 包子1 吃货 yan 吃了 包子2 吃货 yan 吃了 馒头1 吃货 feng 吃了 饺子2 吃货 yan 吃了 馒头2 主进程

View Code