Python - Asyncio模块实现的生产消费者模型

【原创】转载请注明作者Johnthegreat和本文链接

在设计模式中，生产消费者模型占有非常重要的地位，这个模型在现实世界中也有很多有意思的对应场景，比如做包子的人和吃包子的人，当两者速度不匹配时，就需要有一个模型来做匹配（偶合），实现做的包子都会依次消费掉。

import asyncio

class ConsumerProducerModel:

    def __init__(self, producer, consumer, queue=asyncio.Queue(), plate_size=6):  # the plate holds 6pcs bread

        self.queue = queue

        self.producer = producer

        self.consumer = consumer

        self.plate_size = plate_size

    async def produce_bread(self):

        for i in range(self.plate_size):

            bread = f"bread {i}"

            await asyncio.sleep(0.5)  # bread makes faster, 0.5s/pc

            await self.queue.put(bread)

            print(f'{self.producer} makes {bread}')

    async def consume_bread(self):

        while True:

            bread = await self.queue.get()

            await asyncio.sleep(1)  # eat slower, 1s/pc

            print(f'{self.consumer} eats {bread}')

            self.queue.task_done()

async def main():

    queue = asyncio.Queue()

    cp1 = ConsumerProducerModel("John", "Grace", queue)  # group 1

    cp2 = ConsumerProducerModel("Mike", "Lucy", queue)  # group 2

    producer_1 = cp1.produce_bread()

    producer_2 = cp2.produce_bread()

    consumer_1 = asyncio.ensure_future(cp1.consume_bread())

    consumer_2 = asyncio.ensure_future(cp2.consume_bread())

    await asyncio.gather(*[producer_1, producer_2])

    await queue.join()

    consumer_1.cancel()

    consumer_2.cancel()

if __name__ == '__main__':

    loop = asyncio.get_event_loop()

    loop.run_until_complete(main())

    loop.close()

生产消费者模型可以使用多线程和队列来实现，这里选择协程不仅是因为性能不错，而且整个下来逻辑清晰：

1. 先定义初始化的东西，要有个队列，要有生产者，要有消费者，要有装面包的盘子大小；

2. 生产者：根据盘子大小生产出对应的东西（面包），将东西放入盘子（queue）；

3. 消费者：从盘子上取东西，每次取东西都是一个任务，每次任务完成，就标记为task_done（调用函数）。在这个层面，一直循环；

4. 主逻辑：实例化生产消费者模型对象，创建生产者协程，创建任务（ensure_future），收集协程结果，等待所有线程结束（join），手动取消两个消费者协程；

5. 运行：首先创建事件循环，然后进入主逻辑，直到完成，关闭循环。

以上，欢迎交流！

巴特西

Python - Asyncio模块实现的生产消费者模型

最新文章

热门文章