什么是生产者与消费者模式

比如有两个进程A与B，它们共享一个固定大小的缓冲区，A进程生产数据放入缓冲区；B进程从缓冲区取出数据进行计算，那么这里的A进程就相当于生产者，B进程相当于消费者。

为什么要使用生产者与消费者模式

在进程的世界里，生产者就是生产数据的进程，消费者就是使用（处理）数据的进程。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。同样的道理，如果生产者的处理能力大于消费者能力，那么生产者就必须等待消费者。

实现了生产者与消费者的解耦和，平衡了生产力与消费力，因为二者不能直接沟通，而是通过队列进行沟通。

生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者与消费者不直接通信，而是通过阻塞队列进行通信，因此，生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是去阻塞队列中找数据。阻塞队列就类似于缓冲区，平衡了生产者与消费者的能力。

multiprocess-Queue实现

具体代码，如下所示：

from multiprocessing import Process, Queue
import time, random
from threading import Thread
import queue


# 生产者
def producer(name, food, q):
    for i in range(4):
        time.sleep(random.randint(1, 3))  # 模拟产生数据的时间
        f = '%s 生产了 %s %s个' % (name, food, i + 1)
        print(f)
        q.put(f)


# 消费者
def consumer(name, q):
    while True:
        food = q.get()
        if food is None:
            print('%s 获取到一个空' % name)
            break
        f = '%s 消费了 %s' % (name, food)
        print(f)
        time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    q = Queue()  # 创建队列
    # 模拟生产者，产生数据
    p1 = Process(target=producer, args=('p1', '包子', q))
    p1.start()
    p2 = Process(target=producer, args=('p2', '烧饼', q))
    p2.start()

    c1 = Process(target=consumer, args=('c1', q))
    c1.start()
    c2 = Process(target=consumer, args=('c2', q))
    c2.start()

    p1.join()
    p2.join()
    q.put(None)
    q.put(None)

Thread-Queue实现

上面的代码是由多进程实现的，接下来就考虑一下多线程实现该功能。

具体代码，如下所示：

import random
import time
from threading import Thread
import queue


def producer(name, count, q):
    for i in range(count):
        food = f'{name} 生产第{i}个包子'
        print(food)
        q.put(food)


def consumer(name, q):
    while True:
        time.sleep(random.randint(1, 3))
        if q.empty():
            break
        print(f'{name} 消费了 {q.get()}')


if __name__ == '__main__':
    q = queue.Queue()
    print(q.empty())
    for i in range(1, 4):
        p = Thread(target=producer, args=(f'生产者{i}', 10, q))
        p.start()

    for i in range(1, 6):
        c = Thread(target=consumer, args=(f'消费者{i}', q))
        c.start()