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 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.2、yield关键字
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、asyncio
在python3.4及之后的版本。
import asyncio @asyncio.coroutine def func1(): print(1) # 网络IO请求:下载一张图片 yield from asyncio.sleep(2) # 遇到IO耗时操作,自动化切换到tasks中的其他任务 print(2) @asyncio.coroutine def 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 asyncio async 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 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://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 aiohttp import asyncio INSTALL_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(): pass result = 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 asyncio async def func(): print("来玩呀") response = await asyncio.sleep(2) print("结束", response) asyncio.run(func())
示例2:
import asyncio async 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 asyncio async 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 asyncio async 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 task1 ret2 = await task2 print(ret1, ret2) asyncio.run(main())
示例2:
import asyncio async 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 asyncio async 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 asyncio async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务(Future对象),这个任务什么都不干。 fut = loop.create_future() # 等待任务最终结果(Future对象),没有结果则会一直等下去。 await fut asyncio.run(main())
示例2:
import asyncio async 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 fut print(data) asyncio.run(main())
3.5、concurrent.futures.Future对象
使用线程池、进程池实现异步操作时用到的对象。
import time from concurrent.futures import Future from concurrent.futures.thread import ThreadPoolExecutor from concurrent.futures.process import ProcessPoolExecutor def 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 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://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 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(): obj = Reader() async for item in obj: print(item) asyncio.run( func() )
3.8、异步上下文管理器
此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。
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() )
四、uvloop
是asyncio的事件循环的替代方案。事件循环 > 默认asyncio的事件循环。
pip install uvloop
import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # 编写asyncio的代码,与之前写的代码一致。 # 内部的事件循环自动化会变为uvloop asyncio.run(...)
注意:一个asgi -> uvicorn 内部使用的就是uvloop
五、实战案例
5.1、异步redis
在使用python代码操作redis时,链接/操作/断开都是网络IO。
案例1:
pip install aioredis
import asyncio import aioredis async 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 asyncio import aioredis async def execute(address, password): print("开始执行", address) # 网络IO操作:先去连接 47.93.4.197:6379,遇到IO则自动切换任务,去连接47.93.4.198:6379 redis = 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 asyncio import aiomysql async def execute(): # 网络IO操作:连接MySQL conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', 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:
import asyncio import aiomysql async def execute(host, password): print("开始", host) # 网络IO操作:先去连接 47.93.40.197,遇到IO则自动切换任务,去连接47.93.40.198:6379 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('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 asyncio import uvicorn import aioredis from aioredis import Redis from fastapi import FastAPI app = 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 result if __name__ == '__main__': uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")
六、总结
最大的意义:通过一个线程利用其IO等待时间去做一些其他事情。