2022-08-31 09:59:50

一、协程（coroutine）

1.1 协程的概念

**协程也叫“轻量级线程”，是一种用户态中来回切换代码块执行的技术，目的是减少阻塞，提高程序的运行速度。**协程不是计算机提供的功能，而是需要程序员自己实现的，是单个线程内的任务调度技巧。

假设，现在要执行2个函数（不用管谁先完成），这2个函数在运行中各需要3秒阻塞时间去等待网络 IO 的完成：

不使用协程：一个阻塞3秒并执行完成后，另一个才去执行，同样阻塞3秒，总共需要6秒的时间，才能完成两个函数的执行；
使用协程后：先执行的函数，遇到阻塞后，解释器会立马保存阻塞函数的现场数据，并调用另一个函数执行，这样，就相当于同时执行两个函数，总共只需要3秒的时间。大大节省了运行时间，提高了程序的运行效率。

1.2 实现协程的方式

greenlet、gevent 等第三方模块；
yield 关键字；
asyncio 装饰器（python 3.4版本）；
async、await 关键字（python 3.5及以上版本）【推荐】。

二、asyncio 异步编程

2.1 事件循环

事件循环可以理解为一个不断检测并执行代码的死循环，是python 协程系统的核心。它维护着一个任务队列，在整个程序运行过程中，不断循环执行队列中的任务，一旦发生阻塞就切换任务。

import asyncio# python 自带# 获取一个事件循环
loop= asyncio.get_event_loop()# 将任务放到
loop.run_until_complete(任务)

2.2 快速上手

import asyncioasyncdefmain():# 定义协程函数print('hello')await asyncio.sleep(1)print('world')    

asyncio.run(main())# 运行协程函数"""
输出：
hello
world
"""

注意！执行协程函数只会得到协程对象，不会立刻执行函数内的代码。

main()<coroutineobject main at0x1053bb7c8>

2.3 运行协程

要真正运行一个协程，asyncio 提供了三种主要机制：

第一种：用asyncio.run()函数用来运行最高层级的入口点main()函数。（参见上面的示例）

第二种：使用await关键字”等待“一个协程对象（await后面会详解）。以下代码段会在等待 1 秒后打印 “hello”，然后再次等待 2 秒后打印 “world”：

import asyncioimport timeasyncdefsay_after(delay, what):await asyncio.sleep(delay)# 这样才会真正执行 sleep 协程函数print(what)asyncdefmain():print(f"开始：{time.strftime('%X')}")await say_after(1,'hello')await say_after(2,'world')print(f"结束：{time.strftime('%X')}")

asyncio.run(main())

预期的输出:

开始：17:13:52
hello
world
结束：17:13:55

第三种：asyncio.create_task()函数用来并发运行作为 asyncio 任务的多个协程。

让我们修改以上示例，并发运行两个say_after 协程：

asyncdefmain():
    task1= asyncio.create_task(
        say_after(1,'hello'))

    task2= asyncio.create_task(
        say_after(2,'world'))print(f"开始：{time.strftime('%X')}")# 等带两个任务都完成，大约需要2秒await task1await task2print(f"结束：{time.strftime('%X')}")

注意，预期的输出显示代码段的运行时间比之前快了 1 秒:

started at17:14:32
hello
world
finished at17:14:34

2.4 await 关键字

await 可等待对象：表示遇到阻塞后，先挂起当前协程（任务），让事件循环去执行其他任务（如果有其他任务的话），等待“可等待对象”执行完成后，再继续执行下面的代码。

import asyncioasyncdefmain():print('hello')# 会挂起 main 一秒钟，然后打印 world# 一般用于后续步骤需要可等待对象完成后才能执行的情况await asyncio.sleep(1)print('world')

如果可等待对象有返回值，可以直接保存：result = await 可等待对象。

2.5 可等待对象

可等待对象是指可以在await语句中使用的对象，它主要有三种类型：协程、任务和 Future。

2.5.1 协程

在本文中，“协程”可用来表示两个紧密关联的概念：

协程函数：使用async def 函数名定义的函数。
协程对象：调用协程函数得到的对象。

asyncio 也支持旧式的基于生成器（yield 关键字）的协程对象。

2.5.2 任务（Task）

当一个协程被asyncio.create_task()等函数封装成一个任务，该协程就会被自动调度执行：

import asyncioasyncdefnested():return42asyncdefmain():# 创建任务，并将 nested 函数添加到事件循环
    task1= asyncio.create_task(nested())
    task2= asyncio.create_task(nested())# 可以给任务起一个名称# task = asyncio.create_task(nested(), name="t1")# 等待 task 结束await task1await task2

asyncio.run(main())

上面的方法不常用，更加常用的方法是：

import asyncioasyncdefnested():return42asyncdefmain():# 创建任务，并将 nested 函数添加到事件循环
    task_list=[
        asyncio.create_task(nested(),name="t1"),
        asyncio.create_task(nested(),name="t2")]# 等待 task 结束
    done, pending=await asyncio.wait(task_list, timeout=3)# 超时时间是可选的

asyncio.run(main())

说明：

done：所有任务完成后的返回结果的集合。
pending：不常用，任务超时后返回的结果集合。

2.5.3 asyncio.Future

Future 是一个比较底层的可等待对象，任务（Task）是基于 Future 的。Future 一般不会直接用，它表示一个异步操作的最终结果。当一个 Future 对象被等待，这意味着协程将保持等待直到该 Future 对象在其他地方操作完毕。

asyncdefmain():await function_that_returns_a_future_object()# 下面的写法也是有效的await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine())

三、concurrent.futures.Future（补充）

该 Future 对象用于线程池、进程池实现异步操作时用，与 asyncio.Future 没有任何关系，仅仅是名称相同而已。

import timefrom concurrent.futuresimport Futurefrom concurrent.futures.threadimport ThreadPoolExecutorfrom concurrent.futures.processimport ProcessPoolExecutordeffunc(val):
    time.sleep(1)print(val)return"abc"# 创建线程池
pool= ThreadPoolExecutor(max_workers=5)# 创建进程池# pool = ProcessPoolExecutor(max_workers=5)for iinrange(10):
    fut= pool.submit(func, i)# fut 就是 concurrent.futures.Future 对象print(fut)

在实际开发中，可能会出现多进程、多线程和协程交叉实现的情况。比如：基于协程的异步编程 + MySQL（不支持异步）。但我们可以这么做：

import timeimport asyncioimport concurrent.futuresdeffunc1():"""某个耗时操作"""
    time.sleep(2)return"abc"asyncdefmain():# 获取事件循环
    loop=async.get_running_loop()# 1. 在默认的循环执行器中运行
    result=await loop.run_in_executor(None, func1)# 第一个print('default thread pool', result)# 2. 在自定义线程池中运行with concurrent.futures.ThreadPoolExecutor()as pool:
        result=await loop.run_in_executor(
            pool, func1)print('custom thread pool', result)# 3. 在自定义进程池中运行with concurrent.futures.ProcessPoolExecutor()as pool:
        result=await loop.run_in_executor(
            pool, func1)print('custom process pool', result)

asyncio.run(main())

说明：

run_in_executor的参数：

第一个参数是concurrent.futures.Executor实例，如果为None，则使用默认的执行器。
第二个参数就是要执行的函数。

run_in_executor内部做了两件事情：

调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行 func1 函数，并返回一个 concurrent.futures.Future 对象；
调用 asyncio.wrap_future 将 concurrent.futures.Future 对象包装为 asyncio.Future 对象。这样才能使用 await 语法。

3.1 爬虫案例（asyncio+不支持异步的模块）

import asyncioimport requestsasyncdefdownload_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]withopen(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 urlin url_list]
    loop= asyncio.get_event_loop()
    loop.run_until_complete( asyncio.wait(tasks))

四、asyncio 异步迭代器

异步迭代器：
异步迭代器与普通的迭代器是基本一致的，只不过内部实现的是__aiter__()和__anext__()方法。__anext__()必须返回一个awaitable对象。async for会处理异步迭代器的__anext__()方法所返回的可等待对象，知道引发一个StopAsyncIteration异常。
异步可迭代对象：
可以在async for语句中使用的对象，必须通过它的__aiter__()方法返回一个异步迭代器。

举例：

import asyncioclassReader(object):""" 自定义异步迭代器（同时也是异步可迭代对象） """def__init__(self):
        self.count=0asyncdefreadline(self):# await asyncio.sleep(1)
        self.count+=1if self.count==100:returnNonereturn self.countdef__aiter__(self):return selfasyncdef__anext__(self):
        val=await self.readline()if val==None:raise StopAsyncIterationreturn valasyncdeffunc():# 创建异步可迭代对象
    async_iter= Reader()# async for 必须要放在async def函数内，否则语法错误。asyncfor itemin async_iter:print(item)
        
asyncio.run(func())

异步迭代器其实没什么太大的作用，只是支持了async for语法而已。

五、asyncio 异步上下文管理

异步上下文管理需要实现的是__aenter__()和__aexit__()方法，以此实现对async with语句中的环境进行控制。

import asyncioclassAsyncContextManager:def__init__(self):
        self.conn=Noneasyncdefdo_something(self):# 异步操作数据库return666asyncdef__aenter__(self):# 异步链接数据库
        self.conn=await asyncio.sleep(1)return selfasyncdef__aexit__(self, exc_type, exc, tb):# 异步关闭数据库链接await asyncio.sleep(1)asyncdeffunc():# 与异步迭代器一样，必须放在协程函数内asyncwith AsyncContextManager()as f:
        result=await f.do_something()print(result)
        
asyncio.run(func())

六、Uvloop

Python标准库中提供了asyncio模块，用于支持基于协程的异步编程。

uvloop 是 asyncio 中的事件循环的替代方案，替换后可以使得asyncio性能提高。事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言。

安装uvloop：

pipinstall uvloop

在项目中想要使用uvloop替换asyncio的事件循环也非常简单，只要在代码中这么做就行。

import asyncioimport

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

注意：知名的 asgi uvicorn 内部就是使用的uvloop的事件循环。

七、实战案例

7.1 异步Redis

安装 aioredis 模块：

pip3 install aioredis

示例1：异步操作redis，在遇到 IO 等待的地方，使用 await 关键字。

import asyncioimport aioredisasyncdefexecute(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',"root12345"))

示例2：连接多个redis做操作（遇到IO会切换其他任务，提供了性能）。

import asyncioimport aioredisasyncdefexecute(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',"root12345"),
        execute('redis://47.93.4.198:6379',"root12345")]
    
asyncio.run(asyncio.wait(task_list))

更多redis操作参考 aioredis 官网：传送门

7.2 异步MySQL

当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

安装Python异步操作redis模块

pip3 install aiomysql

例子1：

import asyncioimport aiomysqlasyncdefexecute():# 网络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操作：执行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 aiomysqlasyncdefexecute(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.40.197',"root!2345"),
        execute('47.93.40.197',"root!2345")]
   
asyncio.run(asyncio.wait(task_list))

7.3 FastAPI框架

FastAPI 是一款用于构建 API 的高性能 web 框架，框架基于 Python3.6+的type hints搭建。

接下来的异步示例以FastAPI和uvicorn来讲解（uvicorn是一个支持异步的asgi）。

安装 FastAPI：

pip3 install fastapi

安装 uvicorn：

pip3 install uvicorn

举例：

import asyncioimport uvicornimport aioredisfrom aioredisimport Redisfrom fastapiimport FastAPI

app= FastAPI()

REDIS_POOL= aioredis.ConnectionsPool('redis://47.193.14.198:6379', 
                                      password="root123",
                                      minsize=1, 
                                      maxsize=10)

@app.get("/")defindex():""" 普通操作接口"""return{"message":"Hello World"}

@app.get("/red")asyncdefred():""" 异步操作接口"""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 等待过程中去处理其他的请求，提供性能。这就是 FastAPI 如此高性能的原因所在。

python 异步编程：协程与 asyncio

文章目录