python 异步编程：协程与 asyncio

文章目录

• 一、协程（coroutine）
• • 1.1 协程的概念
  • 1.2 实现协程的方式
• 二、asyncio 异步编程
• • 2.1 事件循环
  • 2.2 快速上手
  • 2.3 运行协程
  • 2.4 await 关键字
  • 2.5 可等待对象
  • • 2.5.1 协程
    • 2.5.2 任务（Task）
    • 2.5.3
• 三、（补充）
• • 3.1 爬虫案例（asyncio+不支持异步的模块）
• 四、asyncio 异步迭代器
• 五、asyncio 异步上下文管理
• 六、Uvloop
• 七、实战案例
• • 7.1 异步Redis
  • 7.2 异步MySQL
  • 7.3 FastAPI框架

一、协程（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 提供了三种主要机制：

• 第一种：用()函数用来运行最高层级的入口点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"开始：{('%X')}")
  
      await say_after(1, 'hello')
      await say_after(2, 'world')
  
      print(f"结束：{('%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"开始：{('%X')}")
  
      # 等带两个任务都完成，大约需要2秒
      await task1
      await task2
  
      print(f"结束：{('%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

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

三、（补充）

该 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 就是  对象
    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的参数：

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

run_in_executor内部做了两件事情：

1. 调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行 func1 函数，并返回一个 对象；
2. 调用 asyncio.wrap_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 = [
        '/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        '/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        '/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 (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")

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