# 异步编程-asyncio
为什么要讲?
- 越来越多的学生都来问async异步相关问题,并且这一部分的知识点不太容易学习(异步非阻塞、asyncio)
- 异步相关话题和框架越来越多,例如:tornado、fastapi、django 3.x asgi 、aiohttp都在异步 -> 提升性能。
如何讲解?
- 第一部分:协程。
- 第二部分:asyncio模块进行异步编程。
- 第三部分:实战案例。
# 1.协程
协程不是计算机提供,程序员人为创造。
协程(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) # 第1步:输出 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))
注意:遇到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))
# 2.协程意义
在一个线程中如果遇到IO等待时间,线程不会傻傻等,利用空闲的时候再去干点其他事。
案例:去下载三张图片(网络IO)。
普通方式(同步)
""" pip3 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)
协程方式(异步)
""" 下载图片使用第三方模块aiohttp,请提前安装:pip3 install aiohttp """ #!/usr/bin/env python # -*- coding:utf-8 -*- 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('下载完成',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.异步编程
# 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("come on!")
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 are used to schedule coroutines concurrently.
When a coroutine is wrapped into a Task with functions like
asyncio.create_task()
(opens new window) the coroutine is automatically scheduled to run soon。
白话:在事件循环中添加多个任务的。
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对象
A
Future
is a special low-level awaitable object that represents an eventual result of an asynchronous operation.
Task继承Future,Task对象内部await结果的处理基于Future对象来的。
示例1:
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)
以后写代码可能会存在交叉实现。例如:crm项目80%都是基于协程异步编程 + MySQL(不支持)【线程、进程做异步编程】
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() )
案例: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__()
(opens new window) 和 __anext__()
(opens new window) 方法的对象。__anext__
必须返回一个 awaitable (opens new window) 对象。async for
(opens new window) 会处理异步迭代器的 __anext__()
(opens new window) 方法所返回的可等待对象,直到其引发一个 StopAsyncIteration
(opens new window) 异常。由 PEP 492 (opens new window) 引入。
什么是异步可迭代对象?
可在 async for
(opens new window) 语句中被使用的对象。必须通过它的 __aiter__()
(opens new window) 方法返回一个 asynchronous iterator (opens new window)。由 PEP 492 (opens new window) 引入。
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__()
(opens new window) 和 __aexit__()
(opens new window) 方法来对 async with
(opens new window) 语句中的环境进行控制。由 PEP 492 (opens new window) 引入。
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() )
# 4.uvloop
是asyncio的事件循环的替代方案。事件循环 > 默认asyncio的事件循环。
pip3 install uvloop
import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 编写asyncio的代码,与之前写的代码一致。
# 内部的事件循环自动化会变为uvloop
asyncio.run(...)
注意:一个asgi -> uvicorn
内部使用的就是uvloop
# 5.实战案例
# 5.1 异步redis
在使用python代码操作redis时,连接/操作/断开都是网络IO。
pip3 install aioredis
示例1:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
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://47.93.4.198:6379', "root!2345") )
示例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
pip3 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:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
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框架
安装
pip3 install fastapi
pip3 install uvicorn (asgi内部基于uvloop)
示例: example.py
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI
app = FastAPI()
# 创建一个redis连接池
REDIS_POOL = aioredis.ConnectionsPool('redis://localhost:6379', password="123456", 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("example:app", host="127.0.0.1", port=5000, log_level="info")
# 5.4 爬虫
pip3 install aiohttp
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))
return 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)
if __name__ == '__main__':
asyncio.run( main() )
# 总结
最大的意义:通过一个线程利用其IO等待时间去做一些其他事情。