Python yield关键字实现生产者和消费者

Posted 2023-05-11

参考技术A 通常 yield 用在函数中表示这个函数被定义为一个生成器，可以通过 for 循环去遍历这个生成器对象得到 yield 的值，如下：

yield 除了可以返回值，还能接收调用者传参，下面以 廖雪峰博客
的生产者和消费者为例说明：

若在 c = consumer() 前面加上断点调试可知程序的执行顺序为

python 异步编程

概述

异步编程的基础是协程，

• yield，生成器，借助生成器的特点也可以实现协程代码。
• asyncio，在Python3.4中引入的模块用于编写协程代码。
• async & awiat，在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编写协程代码。

yield

生产者消费者

带yield的函数是一个生成器，不再是一个函数。

def consumer():
    r = 
    n = 0
    while True:
        n = yield r
        if not n:
            return
        print([CONSUMER] Consuming %s... % n)
        r = 200 ok


def produce(c):
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print([PRODUCER] Porducing %s... % n)
        r = c.send(n)
        print([PRODUCER] Consumer return: %s % r)
    c.close()


c = consumer()
produce(c)

async & await

import asyncio
async def func1():
    print(1)
    await asyncio.sleep(2)
    print(2)
async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)
tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

协程意义

通过学习，我们已经了解到协程可以通过一个线程在多个上下文中进行来回切换执行。

但是，协程来回切换执行的意义何在呢？（网上看到很多文章舔协程，协程牛逼之处是哪里呢？）

1. 计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态 反倒会降低性能。
2. IO型的操作，利用协程在IO等待时间就去切换执行其他任务，当IO操作结束后再自动回调，那么就会大大节省资源并提供性能，从而实现异步编程（不等待任务结束就可以去执行其他代码）

案例

"""
下载图片使用第三方模块aiohttp，请提前安装：pip3 install aiohttp
"""
import aiohttp
import asyncio


async def fetch(session, url):
    print("发送请求：", url)
    async with session.get(url, verify_ssl=False) as response:
        content = await response.content.read()
        file_name = url.rsplit(_)[-1]
        with open(file_name, mode=wb) as file_object:
            file_object.write(content)


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg,
            https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg,
            https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg
        ]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)


if __name__ == __main__:
    asyncio.run(main())

异步编程

基于async & await关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术

概念

• 协程函数
• 协程对象
• task对象
• future对象

事件循环

事件循环，可以把他当做是一个while循环，这个while循环在周期性的运行并执行一些任务，在特定条件下终止循环。

import asyncio
loop = asyncio.get_event_loop()

协程和异步编程

协程函数，定义形式为 async def 的函数。

协程对象，调用 协程函数 所返回的对象

定义一个协程函数
async def func():
    pass
# 调用协程函数，返回一个协程对象
result = func()

注意：调用协程函数时，函数内部代码不会执行，只是会返回一个协程对象。

基本使用

程序中，如果想要执行协程函数的内部代码，需要 事件循环 和 协程对象 配合才能实现，如：

import asyncio


async def func():
    pass

result = func()
# 方式一
# loop = asyncio.get_event_loop() # 创建一个事件循环
# loop.run_until_complete(result) # 将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。
# 方式二
# 本质上方式一是一样的，内部先 创建事件循环 然后执行 run_until_complete，一个简便的写法。
# asyncio.run 函数在 Python 3.7 中加入 asyncio 模块，
asyncio.run(result)

这个过程可以简单理解为：将协程当做任务添加到 事件循环 的任务列表，然后事件循环检测列表中的协程是否 已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。

await

await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起 当前协程（任务），当前协程（任务）挂起过程中 事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。代码如下：

import asyncio


async def func1():
  
    print(1)
  	await asyncio.sleep(1)
    print(2)
    
    
async def func2():
  
    print(3)
  	await func1()
    print(4)
 
asyncio.run(func2())

task 对象

Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用 asyncio.create_task() 函数以外，还可以用低层级的 loop.create_task() 或 ensure_future() 函数。不建议手动实例化 Task 对象。

本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。

注意：asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低层级的 asyncio.ensure_future() 函数。

示例一

import asyncio


async def func(p):
    print(%s -- %d % (p, 1))
    await asyncio.sleep(2)
    print(%s -- %d % (p, 2))
    return "返回值"


async def main():
    print("main start")
    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task1 = asyncio.create_task(func(task1))
    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task2 = asyncio.create_task(func(task2))
    print("main end")
    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕并获取结果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)


asyncio.run(main())

示例二

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"


async def main():
    print("main开始")
    # 创建协程，将协程封装到Task对象中并添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    # 在调用
    task_list = [
        asyncio.create_task(func(), name="n1"),
        asyncio.create_task(func(), name="n2")
    ]
    print("main结束")
    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done
    # 如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。
    done, pending = await asyncio.wait(task_list, timeout=None)
    print(done, pending)


asyncio.run(main())

注意：asyncio.wait 源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为[func(),func()] 也是可以的。

示例三

import asyncio


async def func():
    print("执行协程函数内部代码")
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await asyncio.sleep(2)
    print("IO请求结束，结果为：", response)

coroutine_list = [func(), func()]
# 错误：coroutine_list = [asyncio.create_task(func()), asyncio.create_task(func()) ]
# 此处不能直接 asyncio.create_task，因为将Task立即加入到事件循环的任务列表，但此时事件循环还未创建，所以会报错。
# 使用asyncio.wait将列表封装为一个协程，并调用asyncio.run实现执行两个协程
# asyncio.wait内部会对列表中的每个协程执行ensure_future，封装为Task对象。
done, pending = asyncio.run(asyncio.wait(coroutine_list))

总结

• 多个协程函数，协程函数内部调用使用 task_list. = [asyncio.create_task(协程函数) for _ in range(10)]
• 多个协程函数，协程函数外部使用 task_list = [func(), func()] asyncio.await(task_list)
• await asyncio.await(task_list)， 获取执行结果

参照