Python asyncio高性能异步编程 详解

目录

一、协程

1.1、greenlet实现协程

1.2、yield关键字

1.3、asyncio

1.4、async & await关键字

二、协程意义

三、异步编程

3.1、事件循环

3.2、快速上手

3.3、await

3.4、Task对象

3.5、asyncio.Future对象

3.5、concurrent.futures.Future对象

3.7、异步迭代器

3.8、异步上下文管理器

四、uvloop

五、实战案例

5.1、异步redis

5.2、异步MySQL

5.3、FastAPI框架

六、总结

一、协程

协程不是计算机提供，程序员人为创造。

协程（Coroutine），也可以被称为微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块相互切换执行。例如：

def func1():print(1)...print(2)def func2():print(3)...print(4)func1()func2()

实现协程有这么几种方法：

• greenlet，早期模块。

• yield关键字。

• asyncio装饰器（py3.4）

• async、await关键字（py3.5)【推荐】

1.1、greenlet实现协程

pip3 install greenlet

from greenlet import greenletdef func1():print(1)# 第2步：输出 1gr2.switch()# 第3步：切换到 func2 函数print(2)# 第6步：输出 2gr2.switch()# 第7步：切换到 func2 函数，从上一次执行的位置继续向后执行def func2():print(3)# 第4步：输出 3gr1.switch()# 第5步：切换到 func1 函数，从上一次执行的位置继续向后执行print(4)# 第8步：输出 4gr1 = greenlet(func1)gr2 = greenlet(func2)gr1.switch() # 第1步：去执行 func1 函数

1.2、yield关键字

def func1():yield 1yield from func2()yield 2def func2():yield 3yield 4f1 = func1()for item in f1:print(item)

1.3、asyncio

在python3.4及之后的版本。

import asyncio@asyncio.coroutinedef func1():print(1)# 网络IO请求：下载一张图片yield from asyncio.sleep(2)# 遇到IO耗时操作，自动化切换到tasks中的其他任务print(2)@asyncio.coroutinedef func2():print(3)# 网络IO请求：下载一张图片yield from asyncio.sleep(2) # 遇到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))

我们将两个函数放到tasks中，启动时会随机选一个函数执行，遇到IO阻塞自动切换。

1.4、async & await关键字

在python3.5及之后的版本。

import asyncioasync def func1():print(1)# 网络IO请求：下载一张图片await asyncio.sleep(2)# 遇到IO耗时操作，自动化切换到tasks中的其他任务print(2)async def func2():print(3)# 网络IO请求：下载一张图片await asyncio.sleep(2) # 遇到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))

二、协程意义

在一个线程中如果遇到IO等待时间，线程不会傻傻等，利用空闲的时候再去干点其他事。

案例：去下载三张图片（网络IO）。

（1）普通方式（同步）

pip install requests

import requestsdef 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://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']for item in url_list:download_image(item)

（2）协程方式（异步）

pip install aiohttp

import aiohttpimport asyncioINSTALL_AIOHTTP = """pip3 install aiohttp"""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('下载完成', url)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())

三、异步编程

3.1、事件循环

理解成为一个死循环 ，去检测并执行某些代码。

# 伪代码任务列表 = [ 任务1, 任务2, 任务3,... ]while True:可执行的任务列表，已完成的任务列表 = 去任务列表中检查所有的任务，将'可执行'和'已完成'的任务返回for 就绪任务 in 可执行的任务列表:执行已就绪的任务for 已完成的任务 in 已完成的任务列表:在任务列表中移除 已完成的任务如果 任务列表 中的任务都已完成，则终止循环

import asyncio# 去生成或获取一个事件循环loop = asyncio.get_event_loop()# 将任务放到`任务列表`loop.run_until_complete(任务)

3.2、快速上手

协程函数，定义函数时候 async def 函数名 。

协程对象，执行 协程函数() 得到的协程对象。

async def func():passresult = func() # 协程对象

注意：执行协程函数创建协程对象，函数内部代码不会执行。

如果想要运行协程函数内部代码，必须要将协程对象交给事件循环来处理。

import asyncio async def func():print("快来搞我吧！")result = func()# loop = asyncio.get_event_loop()# loop.run_until_complete( result )asyncio.run(result) # python3.7 

3.3、await

await + 可等待的对象（协程对象、Future、Task对象 -> IO等待）

示例1：

import asyncioasync def func():print("来玩呀")response = await asyncio.sleep(2)print("结束", response)asyncio.run(func())

示例2：

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

示例3：

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

await就是等待对象的值得到结果之后再继续向下走，其实就是同步操作。

3.4、Task对象

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

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

示例1：

import asyncioasync def func():print(1)await asyncio.sleep(2)print(2)return "返回值"async def main():print("main开始")# 创建Task对象，将当前执行func函数任务添加到事件循环。task1 = asyncio.create_task(func())# 创建Task对象，将当前执行func函数任务添加到事件循环。task2 = asyncio.create_task(func())print("main结束")# 当执行某协程遇到IO操作时，会自动化切换执行其他任务。# 此处的await是等待相对应的协程全都执行完毕并获取结果ret1 = await task1ret2 = await task2print(ret1, ret2)asyncio.run(main())

示例2：

import asyncioasync def func():print(1)await asyncio.sleep(2)print(2)return "返回值"async def main():print("main开始")task_list = [asyncio.create_task(func(), name='n1'),asyncio.create_task(func(), name='n2')]print("main结束")done, pending = await asyncio.wait(task_list, timeout=None)print(done)asyncio.run(main())

示例3：

import asyncioasync def func():print(1)await asyncio.sleep(2)print(2)return "返回值"task_list = [func(),func(), ]done,pending = asyncio.run(asyncio.wait(task_list))print(done)

3.5、asyncio.Future对象

Task继承Future，Task对象内部await结果的处理基于Future对象来的。

示例1：

import asyncioasync def main():# 获取当前事件循环loop = asyncio.get_running_loop()# 创建一个任务（Future对象），这个任务什么都不干。fut = loop.create_future()# 等待任务最终结果（Future对象），没有结果则会一直等下去。await futasyncio.run(main())

示例2：

import asyncioasync def set_after(fut):await asyncio.sleep(2)fut.set_result("666")async def main():# 获取当前事件循环loop = asyncio.get_running_loop()# 创建一个任务（Future对象），没绑定任何行为，则这个任务永远不知道什么时候结束。fut = loop.create_future()# 创建一个任务（Task对象），绑定了set_after函数，函数内部在2s之后，会给fut赋值。# 即手动设置future任务的最终结果，那么fut就可以结束了。await loop.create_task(set_after(fut) )# 等待 Future对象获取 最终结果，否则一直等下去data = await futprint(data)asyncio.run(main())

3.5、concurrent.futures.Future对象

使用线程池、进程池实现异步操作时用到的对象。

import timefrom concurrent.futures import Futurefrom concurrent.futures.thread import ThreadPoolExecutorfrom concurrent.futures.process import ProcessPoolExecutordef func(value):time.sleep(1)print(value)return 123# 创建线程池pool = ThreadPoolExecutor(max_workers=5)# 创建进程池# pool = ProcessPoolExecutor(max_workers=5)for i in range(10):fut = pool.submit(func, i)print(fut)

案例：asyncio + 不支持异步的模块

import asyncioimport requestsasync def download_image(url):# 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）print("开始下载:", url)loop = asyncio.get_event_loop()# requests模块默认不支持异步操作，所以就使用线程池来配合实现了。future = loop.run_in_executor(None, requests.get, url)response = await futureprint('下载完成')# 图片保存到本地文件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://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 = [download_image(url) for url in url_list]loop = asyncio.get_event_loop()loop.run_until_complete(asyncio.wait(tasks))

3.7、异步迭代器

什么是异步迭代器

实现了 __aiter__() 和 __anext__() 方法的对象。__anext__ 必须返回一个 awaitable 对象。async for 会处理异步迭代器的 __anext__() 方法所返回的可等待对象，直到其引发一个 StopAsyncIteration 异常。由 PEP 492 引入。

什么是异步可迭代对象？

可在 async for 语句中被使用的对象。必须通过它的 __aiter__() 方法返回一个 asynchronous iterator。由 PEP 492 引入。

import asyncioclass Reader(object):""" 自定义异步迭代器（同时也是异步可迭代对象） """def __init__(self):self.count = 0async def readline(self):# await asyncio.sleep(1)self.count += 1if self.count == 100:return Nonereturn self.countdef __aiter__(self):return selfasync def __anext__(self):val = await self.readline()if val == None:raise StopAsyncIterationreturn valasync def func():obj = Reader()async for item in obj:print(item)asyncio.run( func() )

3.8、异步上下文管理器

此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。

import asyncioclass AsyncContextManager:def __init__(self):self.conn = connasync def do_something(self):# 异步操作数据库return 666async def __aenter__(self):# 异步链接数据库self.conn = await asyncio.sleep(1)return selfasync 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() )

四、uvloop

是asyncio的事件循环的替代方案。事件循环 > 默认asyncio的事件循环。

pip install uvloop

import asyncioimport uvloopasyncio.set_event_loop_policy(uvloop.EventLoopPolicy())# 编写asyncio的代码，与之前写的代码一致。# 内部的事件循环自动化会变为uvloopasyncio.run(...)

注意：一个asgi -> uvicorn 内部使用的就是uvloop

五、实战案例

5.1、异步redis

在使用python代码操作redis时，链接/操作/断开都是网络IO。

案例1：

pip install aioredis

import asyncioimport aioredisasync def execute(address, password):print("开始执行", address)# 网络IO操作：创建redis连接redis = await aioredis.create_redis(address, password=password)# 网络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', "123456"))

示例2：

import asyncioimport aioredisasync def execute(address, password):print("开始执行", address)# 网络IO操作：先去连接 47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379redis = await aioredis.create_redis_pool(address, password=password)# 网络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://47.93.4.197:6379', "root!2345"),execute('redis://47.93.4.198:6379', "root!2345")]asyncio.run(asyncio.wait(task_list))

5.2、异步MySQL

pip install aiomysql

示例1：

import asyncioimport aiomysqlasync def execute():# 网络IO操作：连接MySQLconn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', db='mysql', )# 网络IO操作：创建CURSORcur = await conn.cursor()# 网络IO操作：执行SQLawait 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：

import asyncioimport aiomysqlasync def execute(host, password):print("开始", host)# 网络IO操作：先去连接 47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379conn = 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('47.93.41.197', "root!2345"),execute('47.93.40.197', "root!2345")]asyncio.run(asyncio.wait(task_list))

5.3、FastAPI框架

安装

pip install fastapi

pip install uvicorn (asgi内部基于uvloop)

示例： luffy.py

import asyncioimport uvicornimport aioredisfrom aioredis import Redisfrom fastapi import FastAPIapp = FastAPI()# 创建一个redis连接池REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", minsize=1, maxsize=10)@app.get("/")def index():""" 普通操作接口 """# 如果有两个用户并发访问此接口，用户A先执行结束并返回才能用户B执行，同步操作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 resultif __name__ == '__main__':uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")

六、总结

最大的意义：通过一个线程利用其IO等待时间去做一些其他事情。