- 什么是生产者消费者模型 - 基于 Queue 的栗子 - JoinableQueue - 基于 JoinableQueue 的栗子
生产者消费者模型是为了解决数据产生和处理的速度不匹配问题而引入的一种模式。它将负责产生数据的线程或进程称为生产者处理数据的进程或线程称为消费者。
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。他们彼此之间不直接联系，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，而是直接扔给阻塞队列，消费者也不找生产者要数据，而是直接从阻塞队列里取。阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
可以简单的把生产者消费者模型分成三个部分：生产者、队列、消费者。其中生产者和消费者见名即可知意，即别是生产和消费（处理）数据的进程或线程，队列则相当于一个存储数据的仓库，生产者将产生的数据放进去，消费者则从中取出数据，进而处理掉。
我们可以使用 multiprocessing 木块中的 Queue 来构建进程的生产者消费者模型，Queue 与 queue.Queue 类似，也同样是进程和线程安全的。
下面以餐厅的模式为例：厨师是生产者，烹饪食物；顾客是消费者，吃掉食物
from multiprocessing import Process, Queue import time import random import os def consumer(q): while True: res = q.get() # 消费者从仓库中取出数据 time.sleep(random.randint(1, 3)) # 模拟不同的数据处理需要消耗不同的时间 print(' 顾客 {} 吃了 {} '.format(os.getpid(), res)) def producer(q): for i in range(5): time.sleep(random.randint(1, 3)) # 模拟不同的时间点产生了不同的数据 res = '食物{}'.format(i+1) q.put(res) # 生产者将产生的数据放入仓库 print(' 厨师 {} 生产了 {}'.format(os.getpid(), res)) if __name__ == '__main__': q = Queue() p1 = Process(target=producer, args=(q,)) # 生产者进程 c1 = Process(target=consumer, args=(q,)) # 消费者进程 p1.start() c1.start()如果你读了或者运行了上面的代码，就会发现一个问题：消费者的进程似乎一直在运行，不会结束。为什么会这样呢？因为这个生产者太懒了，下班之后通知也不通知消费者一下就自己走了，然后消费者就以为还会有新的数据产生，就一直从仓库里取啊取（因为是while True：）。
那么怎么解决这个问题呢？很简单，让生产者下班的时候告诉一下消费者就可以了。比如往仓库以放一个信号，当消费者取到这个信号时就明白仓库里已经空了，不会再有新的数据了，然后让消费者也走就行了。如下：
# 只需更改一点点 def producer(q): for i in range(5): time.sleep(random.randint(1, 3)) res = '食物{}'.format(i+1) q.put(res) print(' 厨师 {} 生产了 {}'.format(os.getpid(), res)) q.put(None) # 将结束的信号放进去 def consumer(q): while True: res = q.get() if res is None: # 收到信号就结束 break time.sleep(random.randint(1, 3)) print(' 顾客 {} 吃了 {} '.format(os.getpid(), res))你可能会想：为什么要发送信号呢，多麻烦，直接让消费者判断一下仓库是不是空的不就行了，一个简单的 empty() 方法就能搞定。不空就接着处理，空了就走人！像下面
# 这样的话都不用改生产者的代码了 def consumer(q): while not q.empty(): # 如果队列不为空，就取到它空为止 res = q.get() time.sleep(random.randint(1, 3)) print(' 顾客 {} 吃了 {} '.format(os.getpid(), res))事实上这样是不行的！且不说消费者的线程可能先于生产者线程运行，即便是消费者开始运行时仓库不为空也有可能出问题。因为在多进程的环境里，empty() 的结果是不可靠的。如果某一时刻，生产者已经成产了一个数据，但还没来得及放入仓库（即 还没执行q.put(res)），而此时消费者恰好判断仓库为空从而结束了进程。那么这就是不正确的，虽然仓库里暂时没数据了，但还会有新的数据产生。就像顾客吃的速度比厨师做的速度快，顾客吃完盘子里的之后应该等厨师送来新的食物，而不是直接走掉。
你会不会觉的手动传入结束信号很麻烦呢，如果是一对一还好，但要是有多个消费者呢？那你就得传入与消费者个数次结束信号。是很麻烦啊！
multiprocessing 模块提供了一个 JoinableQueue 类，它是 Queue 的子类，额外添加了 task_done() 和 join() 方法。通过这两个方法我们就可以解决这个麻烦了。下面介绍一下 JoinableQueue 的构造函数和这两个方法
JoinableQueue([maxsize]) 功能：创建 JoinableQueue 实例 参数：可选参数 maxsize 表示队列的最大长度，默认为无限制 返回值：一个 JoinableQueue 对象 task_done() 功能：告诉队列之前使用 get() 获取的任务已经完成，该方法由消费者进程使用 参数：无 返回值：None 说明：如果 join() 方法正在阻塞之中，该方法会在所有对象都被处理完的时候返回 (即对之前使用 put() 放进队列中的所有对象都已经返回了对应的 task_done() ) 。 如果被调用的次数多于放入队列中的项目数量，将引发 ValueError 异常 join() 功能：使进程阻塞至队列中所有的元素都被接收和处理完毕 参数：无 返回值：None 说明：当条目（数据任务）被添加到队列时，未完成任务的技术就会加 1 当消费者进程调用 task_done() 方法，表示条目的工作已经完成，未完成技术就会减 1 当未完成计数为 0 时，阻塞才被解除。说完了 JoinableQueue，那就来个栗子品尝一下吧
from multiprocessing import Process, JoinableQueue import time import random import os def consumer(q): while True: res = q.get() time.sleep(random.randint(1,3)) print('顾客{} 吃 {}'.format(os.getpid(), res)) q.task_done() # 向q.join()发送一次信号,证明一个数据已经被取走且处理完了 def producer(name, q): for i in range(2): time.sleep(random.randint(1, 3)) res = '{}{}'.format(name, i+1) q.put(res) print('厨师{} 生产了 {}'.format(os.getpid(), res)) q.join() # 阻塞至所有数据都被处理完 if __name__ == '__main__': q = JoinableQueue() # 生产者们 p1 = Process(target=producer, args=('包子', q)) p2 = Process(target=producer, args=('骨头', q)) p3 = Process(target=producer, args=('汽水', q)) # 消费者们 c1 = Process(target=consumer, args=(q,)) c2 = Process(target=consumer, args=(q,)) c1.daemon = True c2.daemon = True p_lis = [p1, p2, p3, c1, c2] for p in p_lis: p.start() p1.join() p2.join() p3.join() # 问：为什么要把消费者进程设置成守护进程？ # 答：因为p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据，那么c1,c2就没有了存在的价值了,应该随着主进程的结束而结束。