python 构造生产者消费者模型

生产者消费者模型 的建立需要借助第三方进行传递信息。那么使用什么充当这个第三方进行传递信息能够使得生产者消费者模型能够效率更高，实现更为简单呢？

这里使用队列作为这个第三方进行传递信息，连同生产者与消费者。(队列：管道+锁)，既能够传递信息，同时也能够保证数据安全。

普通版

import time
import random
from multiprocessing import Process,Queue

""" 生产者消费者初级模型 """

def producer(name,food,q):
    for i in range(5):
        data = "{}生产了{}{}".format(name,food,i)      # 生产者生产食品
        time.sleep(random.randint(1,3))          # 模拟网络或者其他的延迟，没有什么是处于完全理想状态
        print(data)
        q.put(data)


def consumer(name,q):
    while True:
        food = q.get()          # 从队列中取出数据
        # 判断队列中数据是否已经取出完毕，需要提前在队列中插入表示数据
        if food is None: break
        time.sleep(random.randint(1,3))  # 模拟时间延迟（网络延迟）
        print("{}处理掉了{}".format(name,food))


if __name__ == '__main__':
    q = Queue()
    p1 = Process(target=producer, args=("生产者1", '寿司', q))
    p2 = Process(target=producer, args=("生产者2", '寿司', q))
    c1 = Process(target=consumer, args=("消费者1", q))
    c2 = Process(target=consumer, args=("消费者2", q))
    p1.start()
    p2.start()
    c1.start()
    c2.start()

    # 等待进程运行结束之后，在队列中插入标识生产结束
    p1.join()
    p2.join()
    # 插入的标识的数量需要根据消费者数量进行确定
    q.put(None)
    q.put(None)

这是直接使用多进程里面的模块队列进行传递信息，使得生产者与消费者进行连同，但是这个模型存在一个缺点，那就需要为队列插入特定的结束标识，同时需要确定消费者的数量，插入对应数量的结束标识，同时也需要等待生产者进程运行结束，之后才能插入标识数据，不然会导致进程提前中止。

进阶版

import time
import random
from multiprocessing import Process,Queue,JoinableQueue

""" 生产者消费者进阶模型 """

def producer(name,food,q):
    for i in range(5):
        data = "{}生产了{}{}".format(name,food,i)      # 生产者生产食品
        time.sleep(random.randint(1,3))          # 模拟网络或者其他的延迟，没有什么是处于完全理想状态
        print(data)
        q.put(data)


def consumer(name,q):
    while True:
        food = q.get()          # 从队列中取出数据
        time.sleep(random.randint(1,3))  # 模拟时间延迟（网络延迟）
        print("{}处理掉了{}".format(name,food))
        q.task_done()        # 对队列中的数据的计数减一


if __name__ == '__main__':
    q = JoinableQueue()
    p1 = Process(target=producer, args=("生产者1", '寿司', q))
    p2 = Process(target=producer, args=("生产者2", '寿司', q))
    c1 = Process(target=consumer, args=("消费者1", q))
    c2 = Process(target=consumer, args=("消费者2", q))
    p1.start()
    p2.start()
    # 因为当数据被取出完全的时候，消费者还会在等待取数据，这时候需要守护进程的出现
    c1.daemon = True
    c2.daemon = True
    c1.start()
    c2.start()

    q.join()        # 等待队列中的数据被取出完全

    """ JoinableQueue 这个队列的机制与python的垃圾回收机制中的引用计数相类似 当往队列中插入一个数据时候，计数自动加一， 当调用.task_done()方法使用，对队列的计算进行减一操作； .join()方法与进程中的一样都是阻塞等待，等待队列中的计算清零，才继续运行 """

这样子不但解决了需要设置结束标志，同时也解决了消费者数量的问题。