# asyncio_demo **Repository Path**: paultest/asyncio_demo ## Basic Information - **Project Name**: asyncio_demo - **Description**: 使用Python的asyncio来实现协程/异步编程 - **Primary Language**: Python - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 6 - **Forks**: 4 - **Created**: 2020-06-26 - **Last Updated**: 2024-11-24 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # 协程 & asyncio & 异步编程 ## 1. 协程 想学asyncio，需要先了解协程，协程是根本。 协程不是计算机提供的，是由程序员人为创造的。 协程（Coroutine），也可以称为微线程，是一种用户态内的上下文切换技术，简而言之，其实就是通过一个线程实现代码块相互切换执行，例如： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ def func1(): print(1) ... print(2) def func2(): print(3) ... print(4) func1() func2() ``` 上述代码是普通的函数定义和执行，按流程分别执行两个函数中的代码，并先后输出：1、2、3、4。但如果是介入协程技术的话那么就可以实现函数间代码切换执行，最终输出：1、3、2、4。 在Python中实现协程的方法有： * greenlet：早期模块，第三方模块，用于实现协程代码（Gevent协程就是基于greenlet实现） * yield关键字：生成器，借助生成器的特点也可以实现协程代码 * asyncio装饰器：Python3.4中引入的模块，用于编写协程代码 * async、await关键字：在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编程协程代码【官方推荐】 目前主流使用是Python官方推荐的asyncio模块和async&await关键字结合的方式，例如：在tonado、sanic、fastapi、django3 中均已支持。 ### 1.1 greenlet实现协程 greenlet是一个第三方模块，需要提前安装`pip3 install greenlet`才能使用 ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ from greenlet import greenlet def func1(): print(1) # 第2步：输出1 gr2.switch() # 第3步：切换到 func2 函数 print(2) # 第6步：输出2 gr2.switch() # 第7步：切换到 func2 函数，从上一次执行的位置继续向后执行 def func2(): print(3) # 第4步：输出3 gr1.switch() # 第5步：切换到 func1 函数，从上一次执行的位置继续向后执行 print(4) # 第8步：输出4 gr1 = greenlet(func1) gr2 = greenlet(func2) gr1.switch() # 第1步：去执行 func1 函数 ``` 运行： ``` 1 3 2 4 ``` 注意：switch中也可以传递参数用于在切换执行时相互传递值 ### 1.2 yield关键字 基于Python的生成器的yield和yield form关键字实现协程代码。 ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ def func1(): yield 1 yield from func2() yield 2 def func2(): yield 3 yield 4 f1 = func1() for item in f1: print(item) ``` 运行： ``` 1 3 4 2 ``` 注意：yield from关键字是在Python 3.3中引入的。 ### 1.3 asyncio 在Python3.4之前官方未提供协程的类库，一般大家都是使用greenlet等其他来实现。在Python3.4发布后官方正式支持协程，即：asyncio模块。 ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio @asyncio.coroutine def func1(): print(1) yield from asyncio.sleep(2) # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务 print(2) @asyncio.coroutine def func2(): print(3) yield from asyncio.sleep(2) # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务 print(4) tasks = [ asyncio.ensure_future(func1()), asyncio.ensure_future(func2()) ] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) ``` 运行： ``` 1 3 2 4 ``` 注意：基于asyncio模块实现的协程比之前的要更强，因为它内部还集成了遇到IO耗时操作自动切换的功能。 **注意：Python3.8之后 `@asyncio.coroutine` 装饰器就会被移除，推荐使用async & awit 关键字实现协程代码，即以下方式。** ### 1.4 async & await 关键字 async & awit 关键字在Python3.5版本中正式引入，基于他编写的协程代码其实就是 上一示例 的加强版，让代码可以更加简便。 ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func1(): print(1) await asyncio.sleep(2) # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务 print(2) async def func2(): print(3) await asyncio.sleep(2) # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务 print(4) tasks = [ asyncio.ensure_future(func1()), asyncio.ensure_future(func2()) ] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) ``` 运行： ``` 1 3 2 4 ``` 注意：遇到IO阻塞自动切换到其他任务 ## 2. 协程的意义 在一个线程中如果遇到IO等待时间，线程不会傻傻等待，而是会利用空闲的时间去干其他的事情。 协程可以通过一个线程在多个上下文中进行来回切换执行。 协程来回切换的意义： * 计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态反倒会降低性能。 * IO型的操作，利用协程在IO等待时间就去切换执行其他任务，当IO操作结束后再自动回调，那么就会大大节省资源并提供性能，从而实现异步编程（不等待任务结束就可以去执行其他代码）。 **协程一般应用在有IO操作的程序中**，因为协程可以利用IO等待的时间去执行一些其他的代码，从而提升代码执行效率。 ### 2.1 爬虫案例 案例：下载三张图片 * 普通方式（同步） ```python pip3 install requests ``` ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import requests def download_image(url): print('开始下载：', url) response = requests.get(url) print('下载完成') # 图片保存到本地 file_name = url.rsplit('_')[-1] with open(file_name, mode='wb') as file_object: file_object.write(response.content) if __name__ == '__main__': url_list = [ 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg' ] for item in url_list: download_image(item) ``` * 协程方式（异步） ``` pip3 install aiohttp ``` ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ 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) print('下载完成') async def main(): async with aiohttp.ClientSession() as session: url_list = [ 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg' ] tasks = [asyncio.create_task(fetch(session, url)) for url in url_list] await asyncio.wait(tasks) if __name__ == '__main__': asyncio.run(main()) ``` 上述两种的执行对比之后会发现，基于协程的异步编程 要比 同步编程的效率高了很多。因为： * 同步编程，按照顺序逐一排队执行，如果图片下载时间为2分钟，那么全部执行完则需要6分钟。 * 异步编程，几乎同时发出了3个下载任务的请求（遇到IO请求自动切换去发送其他任务请求），如果图片下载时间为2分钟，那么全部执行完毕也大概需要2分钟左右就可以了。 ## 3. 异步编程 基于`async` & `await`关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术。 ### 3.1 事件循环 事件循环，可以理解为一个while循环，这个while循环在周期性地运行并执行一些任务，在特定条件下终止循环。 ```python # 伪代码 任务列表 = [任务1, 任务2, 任务3, ....] while True: 可执行的任务列表, 已完成的任务列表 = 任务列表中检查所有的任务，将可执行和已完成的任务返回 for 就绪任务 in 可执行的任务列表: 执行已就绪的任务 for 已完成的任务 in 可完成的任务列表: 在任务列表中移除已完成的任务 如果 任务列表 中的任务都已完成，则终止循环 ``` 在编写程序的时候，可以通过如下代码来获取和创建事件循环： ```python import asyncio # 去生成或获取一个事件循环 loop = asyncio.get_event_loop() # 将任务放到 任务列表 loop.run_until_complete(任务) ``` ### 3.2 快速上手 协程函数，定义函数的时候函数名前面加上`async def`，如下： ``` async def 函数名 ``` 协程对象，调用 *协程函数()* 得到的协程对象，比如： ```python # 定义一个协程函数 async def func(): pass # 调用协程函数，返回一个协程对象 result = func() ``` func就是一个协程函数，而result是协程对象 注意：调用协程函数，只是创建协程对象，函数内部代码并不会执行，只是会返回一个协程对象 如果想要运行协程函数内部代码，需要事件循环和协程对象配合才能实现，必须将协程对象交给事件循环来进行处理。 以下是一个简单的事件循环： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func(): print('哈哈哈') # 调用协程函数，返回一个协程对象 result = func() # 去生成一个事件循环 loop = asyncio.get_event_loop() # 将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。 loop.run_until_complete(result) ``` 注意：以上是早期的Python写法，如果是Python 3.7及以后的话，那么会有更简单的写法，如下： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/13 21:33 @Author : boli.hong """ import asyncio async def func(): print('哈哈哈') result = func() # 本质上方式一是一样的，内部先 创建事件循环 然后执行 run_until_complete，一个简便的写法。 asyncio.run(result) ``` 这个过程可以简单理解为：将协程当做任务添加到 事件循环 的任务列表，然后事件循环检测列表中的协程是否 已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。 ### 3.3 await await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起 当前协程（任务），当前协程（任务）挂起过程中 事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。 await + 可等待的对象（IO等待），包括协程对象、Future、Task对象 示例1： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func(): print('哈哈哈哈') # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。 response = await asyncio.sleep(3) print('结束', response) asyncio.run(func()) ``` 示例2： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def others(): print('开始') await asyncio.sleep(3) print('结束') return '返回值' async def func(): print('执行协程函数内部代码') # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务） response = await others() print('IO请求结束，结果为：', response) asyncio.run(func()) ``` 运行： ``` 执行协程函数内部代码 开始 结束 IO请求结束，结果为： 返回值 ``` 示例3： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def others(): print('开始') await asyncio.sleep(3) print('结束') return '返回值' async def func(): print('执行协程函数内部代码') # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务） response1 = await others() print('IO请求结束，结果为：', response1) response2 = await others() print('IO请求结束，结果为：', response2) asyncio.run(func()) ``` **await就是等待对象的值得到结果之后再继续往下走** 上述的所有示例都只是创建了一个任务，即：事件循环的任务列表中只有一个任务，所以在IO等待时无法演示切换到其他任务效果。 在程序想要创建多个任务对象，需要使用Task对象来实现。 ### 3.4 Task对象 > *Tasks* are used to schedule coroutines *concurrently*. > > When a coroutine is wrapped into a *Task* with functions like [`asyncio.create_task()`](https://docs.python.org/3.8/library/asyncio-task.html#asyncio.create_task) the coroutine is automatically scheduled to run soon。 Tasks用于并发调度协程，通过`asyncio.create_task(协程对象)`的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用`asyncio.create_task()`函数以外，还可以用低层级的`loop.create_task()`或`ensure_future()`函数，不建议手动实例化Task对象。 本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。 注意：`asyncio.create_task()`只适用于Python3.7及以上版本，低版本需要使用`loop.create_task()`或`asyncio.ensure_future()` 示例1： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func(): print('开始') await asyncio.sleep(3) print('结束') return '返回值' async def main(): print('main开始') # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态） task1 = asyncio.create_task(func()) task2 = asyncio.create_task(func()) print('main结束') # 当执行某协程遇到IO操作时，会自动化切换执行到其他任务 # 此处的await是等待相应的协程全都执行完毕并获取结果 ret1 = await task1 ret2 = await task2 print(ret1, ret2) asyncio.run(main()) ``` 示例1的用法，用得比较少 示例2： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func(): print('开始') await asyncio.sleep(3) print('结束') return '返回值' async def main(): print('main开始') task_list = [ asyncio.create_task(func()), asyncio.create_task(func()), ] print('main结束') # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。 # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done # 如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。 (done, pending) = await asyncio.wait(task_list) print(done) print(pending) asyncio.run(main()) ``` 注意：Python3.8中的create_task函数新增了name参数，另外，await asyncio.wait返回的结果也和3.7不一样，主要是pending参数，Task字典里面会多了name的属性 注意：`asyncio.wait `源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为`[func(),func()] `也是可以的。 不设置timeout或者是timeout设置为大于3秒 ``` main开始 main结束 开始 开始 结束 结束 { result='返回值'>, result='返回值'>} set() ``` timeout设置为1-2秒 ``` main开始 main结束 开始 开始 set() { wait_for=()]>>, wait_for=()]>>} ``` 没有打印出结束 timeout设置为3秒 ``` main开始 main结束 开始 开始 set() { wait_for=>, wait_for=>} 结束 结束 ``` 示例3： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def func(): print('开始') await asyncio.sleep(3) print('结束') return '返回值' task_list = [ func(), func(), ] # 错误：task_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(task_list)) print(done) print(pending) ``` 运行： ``` 开始 开始 结束 结束 { result='返回值'>, result='返回值'>} set() ``` ### 3.5 asyncio.Future对象 > A `Future`is a special **low-level** awaitable object that represents an **eventual result** of an asynchronous operation. asyncio中的Future对象是一个相对更偏向底层的可对象，通常我们不会直接用到这个对象，而是直接使用Task对象来完成任务的并和状态的追踪。（ Task 是 Futrue的子类 ） Future为我们提供了异步编程中的 最终结果 的处理（Task类也具备状态处理的功能）。 示例1： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务（Future对象，如果没绑定任何行为，则这个任务永远不会结束） future = loop.create_future() # 等待Future对象获取最终结果，否则会一直等下去 await future asyncio.run(main()) ``` 示例2： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio async def set_after(future): await asyncio.sleep(2) future.set_result("666") async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务（Future对象，如果没绑定任何行为，则这个任务永远不会结束） future = loop.create_future() # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2秒之后会给future赋值，即手动设置future任务的最终结果，那么future就可以结束了 await loop.create_task(set_after(future)) # 等待Future对象获取最终结果，否则会一直等下去 data = await future print(data) asyncio.run(main()) ``` 运行： 666 Future对象本身函数进行绑定，所以想要让事件循环获取Future的结果，则需要手动设置。而Task对象继承了Future对象，其实就对Future进行扩展，他可以实现在对应绑定的函数执行完成之后，自动执行`set_result`，从而实现自动结束。 虽然，平时使用的是Task对象，但对于结果的处理本质是基于Future对象来实现的。 扩展：支持 `await 对象`语 法的对象课成为可等待对象，所以 `协程对象`、`Task对象`、`Future对象` 都可以被成为可等待对象。 ### 3.6 futures.Future对象 在Python的`concurrent.futures`模块中也有一个Future对象，这个对象是基于线程池和进程池实现异步操作时使用的对象。 使用线程池、进程池来实现异步操作时用到的对象，即使用线程、进程做异步编程 两个Future对象是不同的，他们是为不同的应用场景而设计，例如：`concurrent.futures.Future`不支持await语法 等。 官方提示两对象之间不同： - unlike asyncio Futures, [`concurrent.futures.Future`](https://docs.python.org/3.8/library/concurrent.futures.html#concurrent.futures.Future) instances cannot be awaited. - [`asyncio.Future.result()`](https://docs.python.org/3.8/library/asyncio-future.html#asyncio.Future.result) and [`asyncio.Future.exception()`](https://docs.python.org/3.8/library/asyncio-future.html#asyncio.Future.exception) do not accept the *timeout* argument. - [`asyncio.Future.result()`](https://docs.python.org/3.8/library/asyncio-future.html#asyncio.Future.result) and [`asyncio.Future.exception()`](https://docs.python.org/3.8/library/asyncio-future.html#asyncio.Future.exception) raise an [`InvalidStateError`](https://docs.python.org/3.8/library/asyncio-exceptions.html#asyncio.InvalidStateError) exception when the Future is not *done*. - Callbacks registered with [`asyncio.Future.add_done_callback()`](https://docs.python.org/3.8/library/asyncio-future.html#asyncio.Future.add_done_callback) are not called immediately. They are scheduled with [`loop.call_soon()`](https://docs.python.org/3.8/library/asyncio-eventloop.html#asyncio.loop.call_soon) instead. - asyncio Future is not compatible with the [`concurrent.futures.wait()`](https://docs.python.org/3.8/library/concurrent.futures.html#concurrent.futures.wait) and [`concurrent.futures.as_completed()`](https://docs.python.org/3.8/library/concurrent.futures.html#concurrent.futures.as_completed) functions. 在Python提供了一个将`futures.Future` 对象包装成`asyncio.Future`对象的函数 `asynic.wrap_future`。 一般在程序开发中我们要么统一使用 asycio 的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果 `协程的异步`和 `进程池/线程池的异步` 混搭时，那么就会用到此功能了 注意：比起使用asyncio，使用线程、进程来做异步编程的话，耗费的资源会更高，这是因为协程是代码自己在单个线程中创建的，而进程、线程是操作系统本身支持的，切换进行/线程是需要进行操作系统级别的上下文切换，所需要的资源会更高，**一般来说，优先使用协程，其次是线程，最后才是进程** ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import time from concurrent.futures import Future from concurrent.futures.thread import ThreadPoolExecutor from concurrent.futures.process import ProcessPoolExecutor def func(value): print('func start：', time.strftime('%H:%M:%S', time.localtime(time.time()))) time.sleep(3) print(value) print('func end：', time.strftime('%H:%M:%S', time.localtime(time.time()))) print("----------\r\n") return value * 10 # 创建线程池，最多为5个线程 pool = ThreadPoolExecutor(max_workers=5) # 或者是以下写法，创建进程池 # pool = ProcessPoolExecutor(max_workers=5) for i in range(10): print('start：', time.strftime('%H:%M:%S', time.localtime(time.time()))) # 使用线程池执行函数 fut = pool.submit(func, i) print('result：', fut.result) print('end：', time.strftime('%H:%M:%S', time.localtime(time.time()))) print("-----------------\r\n") ``` 运行： ``` start： 23:04:48 func start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 func start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 func start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 func start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 func start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 result： > end： 23:04:48 ----------------- start： 23:04:48 result： > end： 23:04:48 ----------------- 0 func end： 23:04:51 ---------- func start： 23:04:51 1 func end： 2 func end： 23:04:51 ---------- 23:04:51 ---------- func start： 23:04:51 func start： 23:04:51 3 func end： 23:04:51 ---------- func start： 23:04:51 4 func end： 23:04:51 ---------- func start： 23:04:51 5 func end： 23:04:54 ---------- 7 func end： 68 func end： 23:04:54 23:04:54 ---------- 9 func end： 23:04:54 func end： 23:04:54 ---------- ---------- ---------- ``` 可以看出： * 使用线程池来执行10次func函数，23:04:48在time.sleep之前就全部打印出来了，Future对象的result的状态为：pending * sleep之后，分为两个时间进行打印，分别是23:04:51和23:04:54，这是因为我们只开了5个线程，而有10个任务，所以需要分为两次执行 * 注意：sleep之后，打印的结果的顺序并不是绝对的，比如func end：68这一行，是因为有两个进程同时返回了结果，同时打印在了控制台，导致打印的结果是乱的 该对象的使用场景： 在异步编程中，有一些组件或第三方模块不支持协程，比如说requests、MySQL等，不支持协程，所以遇到这种情况需要使用线程、进程来做异步编程 比如：CRM项目中80%都是基于协程异步编程+MySQL ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import time import asyncio import concurrent.futures def func1(): # 某个耗时操作 time.sleep(2) return "SB" async def main(): loop = asyncio.get_running_loop() # 1. Run in the default loop's executor ( 默认ThreadPoolExecutor ) # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象 # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。 # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。 fut = loop.run_in_executor(None, func1) result = await fut print('default thread pool', result) # 2. Run in a custom thread pool: # with concurrent.futures.ThreadPoolExecutor() as pool: # result = await loop.run_in_executor( # pool, func1) # print('custom thread pool', result) # 3. Run in a custom process pool: # with concurrent.futures.ProcessPoolExecutor() as pool: # result = await loop.run_in_executor( # pool, func1) # print('custom process pool', result) asyncio.run(main()) ``` 运行： ``` default thread pool SB ``` 代码中的`fut = loop.run_in_executor(None, func1)`就相当于： ```python pool = ThreadPoolExecutor() fut = pool.submit(func1) ``` 由于返回的是`concurrent.futures.Future`对象，该对象不支持await语法，所以需要包装为asycio.Future对象，即`result = await fut` 另一种写法为： ```python loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor( pool, func1) print('custom thread pool', result) ``` 另外，也可以使用进程： ```python loop = asyncio.get_running_loop() with concurrent.futures.ProcessPoolExecutor() as pool: result = await loop.run_in_executor( pool, func1) print('custom process pool', result) ``` 案例：asyncio + 不支持异步的模块，比如requests ```python # -*- encoding: utf-8 -*- """ @Time : 2020/6/25 @Author : hongboli """ import asyncio import requests async def download_image(url): # 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务） print("开始下载:", url) loop = asyncio.get_event_loop() # requests模块默认不支持异步操作，所以就使用线程池来配合实现了。 future = loop.run_in_executor(None, requests.get, url) response = await future print('下载完成') # 图片保存到本地文件 file_name = url.rsplit('_')[-1] with open(file_name, mode='wb') as file_object: file_object.write(response.content) if __name__ == '__main__': url_list = [ 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg', 'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg' ] tasks = [download_image(url) for url in url_list] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) ``` 运行： ``` 开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg 开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg 开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg 下载完成 下载完成 下载完成 ``` ### 3.7 异步迭代器 **什么是异步迭代器** 实现了 [`__aiter__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__aiter__) 和 [`__anext__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__anext__) 方法的对象。`__anext__` 必须返回一个 [awaitable](https://docs.python.org/zh-cn/3.8/glossary.html#term-awaitable) 对象。[`async for`](https://docs.python.org/zh-cn/3.8/reference/compound_stmts.html#async-for) 会处理异步迭代器的 [`__anext__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__anext__) 方法所返回的可等待对象，直到其引发一个 [`StopAsyncIteration`](https://docs.python.org/zh-cn/3.8/library/exceptions.html#StopAsyncIteration) 异常。由 [**PEP 492**](https://www.python.org/dev/peps/pep-0492) 引入。 **什么是异步可迭代对象？** 可在 [`async for`](https://docs.python.org/zh-cn/3.8/reference/compound_stmts.html#async-for) 语句中被使用的对象。必须通过它的 [`__aiter__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__aiter__) 方法返回一个 [asynchronous iterator](https://docs.python.org/zh-cn/3.8/glossary.html#term-asynchronous-iterator)。由 [**PEP 492**](https://www.python.org/dev/peps/pep-0492) 引入。 ```python import asyncio class Reader(object): """ 自定义异步迭代器（同时也是异步可迭代对象） """ def __init__(self): self.count = 0 async def readline(self): # await asyncio.sleep(1) self.count += 1 if self.count == 100: return None return self.count def __aiter__(self): return self async def __anext__(self): val = await self.readline() if val == None: raise StopAsyncIteration return val async def func(): # 创建异步可迭代对象 async_iter = Reader() # async for 必须要放在async def函数内，否则语法错误。 async for item in async_iter: print(item) asyncio.run(func()) ``` 异步迭代器其实没什么太大的作用，只是支持了async for语法而已。 ### 3.8 异步上下文管理 此种对象通过定义 [`__aenter__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__aenter__) 和 [`__aexit__()`](https://docs.python.org/zh-cn/3.8/reference/datamodel.html#object.__aexit__) 方法来对 [`async with`](https://docs.python.org/zh-cn/3.8/reference/compound_stmts.html#async-with) 语句中的环境进行控制。由 [**PEP 492**](https://www.python.org/dev/peps/pep-0492) 引入。 ```python import asyncio class AsyncContextManager: def __init__(self): self.conn = conn async def do_something(self): # 异步操作数据库 return 666 async def __aenter__(self): # 异步链接数据库 self.conn = await asyncio.sleep(1) return self async def __aexit__(self, exc_type, exc, tb): # 异步关闭数据库链接 await asyncio.sleep(1) async def func(): async with AsyncContextManager() as f: result = await f.do_something() print(result) asyncio.run(func()) ``` 只要对象里面定义了`__aenter__`，就可以使用`async with 对象实例 as f`，当然，同样和上个例子一样，async with同样是需要放在async def函数内的 这个异步的上下文管理器还是比较有用的，平时在开发过程中 打开、处理、关闭 操作时，就可以用这种方式来处理。 ### 3.9 小结 在程序中只要看到`async`和`await`关键字，其内部就是基于协程实现的异步编程，这种异步编程是通过一个线程在IO等待时间去执行其他任务，从而实现并发。 以上就是异步编程的常见操作，内容参考官方文档。 - 中文版：https://docs.python.org/zh-cn/3.8/library/asyncio.html - 英文本：https://docs.python.org/3.8/library/asyncio.html ## 4. uvloop Python标准库中提供了`asyncio`模块，用于支持基于协程的异步编程。 uvloop是 asyncio 中的事件循环的替代方案，替换后可以使得asyncio性能提高。事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言。 安装uvloop ``` pip3 install uvloop ``` 在项目中想要使用uvloop替换asyncio的事件循环也非常简单，只要在代码中这么做就行。 ```python import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # 编写asyncio的代码，与之前写的代码一致。 # 内部的事件循环自动化会变为uvloop asyncio.run(...) ``` 注意：知名的asgi uvicorn内部就是使用的uvloop的事件循环。 ## 5.实战案例 为了更好理解，上述所有示例的IO情况都是以 `asyncio.sleep` 为例，而真实的项目开发中会用到很多IO的情况。 ### 5.1 异步redis 当通过python去操作redis时，链接、设置值、获取值 这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。 在使用python代码操作redis时，链接/操作/断开都是网络IO。 示例1：异步操作redis： 安装Python异步操作redis模块 ``` pip3 install aioredis ``` ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import asyncio import aioredis async def execute(address, password=None): print("开始执行", address) # 网络IO操作：创建redis连接 if password: redis = await aioredis.create_redis(address, password=password) else: redis = await aioredis.create_redis(address) # 网络IO操作：在redis中设置哈希值car，内部在设三个键值对，即： redis = { car:{key1:1,key2:2,key3:3}} await redis.hmset_dict('car', key1=1, key2=2, key3=3) # 网络IO操作：去redis中获取值 result = await redis.hgetall('car', encoding='utf-8') print(result) redis.close() # 网络IO操作：关闭redis连接 await redis.wait_closed() print("结束", address) asyncio.run(execute('redis://127.0.0.1:6379')) ``` 示例2：连接多个redis做操作（遇到IO会切换其他任务，提供了性能）： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import asyncio import aioredis async def execute(address, password=None): print("开始执行", address) # 网络IO操作：先去连接第一个 ，遇到IO则自动切换任务，去连接另一个 if password: redis = await aioredis.create_redis_pool(address, password=password) else: redis = await aioredis.create_redis_pool(address) # 网络IO操作：遇到IO会自动切换任务 await redis.hmset_dict('car', key1=1, key2=2, key3=3) # 网络IO操作：遇到IO会自动切换任务 result = await redis.hgetall('car', encoding='utf-8') print(result) redis.close() # 网络IO操作：遇到IO会自动切换任务 await redis.wait_closed() print("结束", address) task_list = [ execute('redis://127.0.0.1:6379'), execute('redis://127.0.0.1:6379') ] asyncio.run(asyncio.wait(task_list)) ``` 更多redis操作参考aioredis官网：https://aioredis.readthedocs.io/en/v1.3.0/start.html ### 5.2 异步MySQL 当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。 安装Python异步操作redis模块 ``` pip3 install aiomysql ``` 示例1： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import asyncio import aiomysql async def execute(): # 网络IO操作：连接MySQL conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123456', db='mysql', ) # 网络IO操作：创建CURSOR cur = await conn.cursor() # 网络IO操作：执行SQL await cur.execute("SELECT Host,User FROM user") # 网络IO操作：获取SQL结果 result = await cur.fetchall() print(result) # 网络IO操作：关闭链接 await cur.close() conn.close() asyncio.run(execute()) ``` 示例2： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import asyncio import aiomysql async def execute(host, password): print("开始", host) # 网络IO操作：先去连接 127.0.0.1，遇到IO则自动切换任务，去连接另一个127.0.0.1 conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql') # 网络IO操作：遇到IO会自动切换任务 cur = await conn.cursor() # 网络IO操作：遇到IO会自动切换任务 await cur.execute("SELECT Host,User FROM user") # 网络IO操作：遇到IO会自动切换任务 result = await cur.fetchall() print(result) # 网络IO操作：遇到IO会自动切换任务 await cur.close() conn.close() print("结束", host) task_list = [ execute('127.0.0.1', "123456"), execute('127.0.0.1', "123456") ] asyncio.run(asyncio.wait(task_list)) ``` ### 5.3 FastApi框架 FastAPI是一款用于构建API的高性能web框架，框架基于Python3.6+的 `type hints`搭建。 接下里的异步示例以`FastAPI`和`uvicorn`来讲解（uvicorn是一个支持异步的asgi）。 安装FastAPI web 框架， ``` pip3 install fastapi ``` 安装uvicorn，本质上为web提供socket server的支持的asgi（一般支持异步称asgi、不支持异步称wsgi） ``` pip3 install uvicorn # 内部是基于uvloop ``` 示例： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import asyncio import uvicorn import aioredis from aioredis import Redis from fastapi import FastAPI app = FastAPI() # 创建redis的连接池 REDIS_POOL = aioredis.ConnectionsPool('redis://127.0.0.1:6379', minsize=1, maxsize=10) @app.get("/") def index(): """ 普通操作接口 """ return {"message": "Hello World"} @app.get("/red") async def red(): """ 异步操作接口 """ print("请求来了") await asyncio.sleep(3) # 连接池获取一个连接 conn = await REDIS_POOL.acquire() redis = Redis(conn) # 设置值 await redis.hmset_dict('car', key1=1, key2=2, key3=3) # 读取值 result = await redis.hgetall('car', encoding='utf-8') print(result) # 连接归还连接池 REDIS_POOL.release(conn) return result if __name__ == '__main__': # 注意：第一个参数是文件名 uvicorn.run("5.3_fastapi:app", host="127.0.0.1", port=5000, log_level="info") ``` 在有多个用户并发请求的情况下，异步方式来编写的接口可以在IO等待过程中去处理其他的请求，提供性能。 例如：同时有两个用户并发来向接口 `http://127.0.0.1:5000/red` 发送请求，服务端只有一个线程，同一时刻只有一个请求被处理。 异步处理可以提供并发是因为：当视图函数在处理第一个请求时，第二个请求此时是等待被处理的状态，当第一个请求遇到IO等待时，会自动切换去接收并处理第二个请求，当遇到IO时自动化切换至其他请求，一旦有请求IO执行完毕，则会再次回到指定请求向下继续执行其功能代码。 ### 5.4 爬虫 在编写爬虫应用时，需要通过网络IO去请求目标数据，这种情况适合使用异步编程来提升性能，接下来我们使用支持异步编程的aiohttp模块来实现。 安装aiohttp模块 ``` pip3 install aiohttp ``` 示例： ```python # -*- encoding: utf-8 -*- """ @Time : 2020/7/12 @Author : hongboli """ import aiohttp import asyncio async def fetch(session, url): print("发送请求：", url) async with session.get(url, verify_ssl=False) as response: text = await response.text() print("得到结果：", url, len(text)) async def main(): async with aiohttp.ClientSession() as session: url_list = [ 'https://python.org', 'https://www.baidu.com', 'https://www.pythonav.com' ] tasks = [asyncio.create_task(fetch(session, url)) for url in url_list] (done, pending) = await asyncio.wait(tasks) print(done) print(pending) if __name__ == '__main__': asyncio.run(main()) ``` ## 总结 最大的意义：通过一个线程利用其IO等待时间去做一些其他事情。 为了提升性能越来越多的框架都在向异步编程靠拢，例如：sanic、tornado、django3.0、django channels组件 等，用更少资源可以做处理更多的事，何乐而不为呢。