你好赵伟

这个屌丝很懒，什么也没留下！

热门标签

python 异步编程：协程与 asyncio_python协程asyncio

作者：你好赵伟 | 2024-02-10 18:39:28

踩

python协程asyncio

一、协程（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(任务)
1
2
3
4
5
6

2.2 快速上手

import asyncio

async def main():   # 定义协程函数
    print('hello')    
    await asyncio.sleep(1)
    print('world')    

asyncio.run(main())    # 运行协程函数
"""
输出：
hello
world
"""
1
2
3
4
5
6
7
8
9
10
11
12
13

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

main()
<coroutine object main at 0x1053bb7c8>
1
2

2.3 运行协程

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

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

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

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)   # 这样才会真正执行 sleep 协程函数
    print(what)

async def main():
    print(f"开始：{time.strftime('%X')}")

    await say_after(1, 'hello')
    await say_after(2, 'world')

    print(f"结束：{time.strftime('%X')}")

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

预期的输出:

开始：17:13:52
hello
world
结束：17:13:55
1
2
3
4

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

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

async def main():
    task1 = asyncio.create_task(
        say_after(1, 'hello'))

    task2 = asyncio.create_task(
        say_after(2, 'world'))

    print(f"开始：{time.strftime('%X')}")

    # 等带两个任务都完成，大约需要2秒
    await task1
    await task2

    print(f"结束：{time.strftime('%X')}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14

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

started at 17:14:32
hello
world
finished at 17:14:34
1
2
3
4

2.4 await 关键字

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

import asyncio

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

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

2.5 可等待对象

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

2.5.1 协程

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

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

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

2.5.2 任务（Task）

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

import asyncio

async def nested():
    return 42

async def main():
    # 创建任务，并将 nested 函数添加到事件循环 
    task1 = asyncio.create_task(nested())
    task2 = asyncio.create_task(nested())
    # 可以给任务起一个名称
    # task = asyncio.create_task(nested(), name="t1")

    # 等待 task 结束
    await task1
    await task2

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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

import asyncio

async def nested():
    return 42

async def main():
    # 创建任务，并将 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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

说明：

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

2.5.3 asyncio.Future

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

async def main():
    await function_that_returns_a_future_object()

    # 下面的写法也是有效的
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )
1
2
3
4
5
6
7
8

三、concurrent.futures.Future（补充）

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

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor

def func(val):
    time.sleep(1)
    print(val)
    return "abc"

# 创建线程池
pool = ThreadPoolExecutor(max_workers=5)

# 创建进程池
# pool = ProcessPoolExecutor(max_workers=5)

for i in range(10):
    fut = pool.submit(func, i)    
    # fut 就是 concurrent.futures.Future 对象
    print(fut)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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

import time
import asyncio
import concurrent.futures

def func1():
    """某个耗时操作"""
    time.sleep(2)
    return "abc"

async def main():
    # 获取事件循环
    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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

说明：

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 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) )
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

四、asyncio 异步迭代器

异步迭代器：

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

可以在async for语句中使用的对象，必须通过它的__aiter__()方法返回一个异步迭代器。

举例：

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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

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

五、asyncio 异步上下文管理

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

import asyncio

class AsyncContextManager:
    def __init__(self):
        self.conn = None
        
    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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

六、Uvloop

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

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

安装uvloop：

pip install uvloop
1

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

import asyncioimport

uvloopasyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 编写asyncio的代码，与之前写的代码一致。

# 内部的事件循环自动化会变为
uvloopasyncio.run(...)
1
2
3
4
5
6
7
8

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

七、实战案例

7.1 异步Redis

安装 aioredis 模块：

pip3 install aioredis
1

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

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', "root12345"))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

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

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', "root12345"),
        execute('redis://47.93.4.198:6379', "root12345")
    ]
    
asyncio.run(asyncio.wait(task_list))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

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

7.2 异步MySQL

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

安装Python异步操作redis模块

pip3 install aiomysql
1

例子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())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

例子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.40.197', "root!2345"),
        execute('47.93.40.197', "root!2345")
    ]
   
asyncio.run(asyncio.wait(task_list))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

7.3 FastAPI框架

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

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

安装 FastAPI：

pip3 install fastapi
1

安装 uvicorn：

pip3 install uvicorn
1

举例：

import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI

app = FastAPI()

REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', 
                                      password="root123",
                                      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("luffy:app", host="127.0.0.1", port=5000, log_level="info")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

在有多个用户并发请求的情况下，异步方式来编写的接口可以在 IO 等待过程中去处理其他的请求，提供性能。这就是 FastAPI 如此高性能的原因所在。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/74987