开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Python知识库 -> python 异步编程：协程与 asyncio -> 正文阅读

[Python知识库]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(任务)

2.2 快速上手

import asyncio

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

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

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

main()
<coroutine object main at 0x1053bb7c8>

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())

预期的输出:

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

第三种：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 秒:

started at 17:14:32
hello
world
finished at 17:14:34

2.4 await 关键字

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

import asyncio

async def main():
    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 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())

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

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())

说明：

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()
    )

三、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)

在实际开发中，可能会出现多进程、多线程和协程交叉实现的情况。比如：基于协程的异步编程 + 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())

说明：

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) )

四、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())

异步迭代器其实没什么太大的作用，只是支持了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())

六、Uvloop

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

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

安装uvloop：

pip install 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 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"))

示例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))

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

7.2 异步MySQL

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

安装Python异步操作redis模块

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：

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))

7.3 FastAPI框架

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

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

安装 FastAPI：

pip3 install fastapi

安装 uvicorn：

pip3 install uvicorn

举例：

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")