协程 & asyncio & 异步编程

1. 协程

想学asyncio，需要先了解协程，协程是根本。

协程不是计算机提供的，是由程序员人为创造的。协程（Coroutine），也可以称为微线程，是一种用户态内的上下文切换技术，简而言之，其实就是通过一个线程实现代码块相互切换执行，例如：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""


def func1():
	print(1)
    ...
	print(2)

def func2():
	print(3)
    ...
	print(4)

func1()
func2()

上述代码是普通的函数定义和执行，按流程分别执行两个函数中的代码，并先后输出：1、2、3、4。但如果是介入协程技术的话那么就可以实现函数间代码切换执行，最终输出：1、3、2、4。

在Python中实现协程的方法有：

greenlet：早期模块，第三方模块，用于实现协程代码（Gevent协程就是基于greenlet实现）
yield关键字：生成器，借助生成器的特点也可以实现协程代码
asyncio装饰器：Python3.4中引入的模块，用于编写协程代码
async、await关键字：在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编程协程代码【官方推荐】

目前主流使用是Python官方推荐的asyncio模块和async&await关键字结合的方式，例如：在tonado、sanic、fastapi、django3 中均已支持。

1.1 greenlet实现协程

greenlet是一个第三方模块，需要提前安装pip3 install greenlet才能使用

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
from greenlet import greenlet


def func1():
	print(1)       # 第2步：输出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 函数

运行：

注意：switch中也可以传递参数用于在切换执行时相互传递值

1.2 yield关键字

基于Python的生成器的yield和yield form关键字实现协程代码。

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""


def func1():
	yield 1
	yield from func2()
	yield 2

def func2():
	yield 3
	yield 4

f1 = func1()
for item in f1:
	print(item)

运行：

注意：yield from关键字是在Python 3.3中引入的。

1.3 asyncio

在Python3.4之前官方未提供协程的类库，一般大家都是使用greenlet等其他来实现。在Python3.4发布后官方正式支持协程，即：asyncio模块。

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


@asyncio.coroutine
def func1():
	print(1)
	yield from asyncio.sleep(2)    # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务
	print(2)

@asyncio.coroutine
def func2():
	print(3)
	yield from asyncio.sleep(2)    # 遇到IO耗时操作，比如说网络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))

运行：

注意：基于asyncio模块实现的协程比之前的要更强，因为它内部还集成了遇到IO耗时操作自动切换的功能。

注意：Python3.8之后 @asyncio.coroutine 装饰器就会被移除，推荐使用async & awit 关键字实现协程代码，即以下方式。

1.4 async & await 关键字

async & awit 关键字在Python3.5版本中正式引入，基于他编写的协程代码其实就是上一示例的加强版，让代码可以更加简便。

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


async def func1():
	print(1)
	await asyncio.sleep(2)    # 遇到IO耗时操作，比如说网络IO请求等，自动化切换到tasks中的其他任务
	print(2)

async def func2():
	print(3)
	await asyncio.sleep(2)    # 遇到IO耗时操作，比如说网络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阻塞自动切换到其他任务

2. 协程的意义

在一个线程中如果遇到IO等待时间，线程不会傻傻等待，而是会利用空闲的时间去干其他的事情。

协程可以通过一个线程在多个上下文中进行来回切换执行。

协程来回切换的意义：

计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态反倒会降低性能。
IO型的操作，利用协程在IO等待时间就去切换执行其他任务，当IO操作结束后再自动回调，那么就会大大节省资源并提供性能，从而实现异步编程（不等待任务结束就可以去执行其他代码）。

协程一般应用在有IO操作的程序中，因为协程可以利用IO等待的时间去执行一些其他的代码，从而提升代码执行效率。

2.1 爬虫案例

案例：下载三张图片

普通方式（同步）

pip3 install requests

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
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://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg',
        'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg',
        'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg'
    ]
    
    for item in url_list:
    	download_image(item)

协程方式（异步）

pip3 install aiohttp

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
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('下载完成')


async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg',
            'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg',
            'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg'
        ]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)


if __name__ == '__main__':
    asyncio.run(main())

上述两种的执行对比之后会发现，基于协程的异步编程要比同步编程的效率高了很多。因为：

同步编程，按照顺序逐一排队执行，如果图片下载时间为2分钟，那么全部执行完则需要6分钟。
异步编程，几乎同时发出了3个下载任务的请求（遇到IO请求自动切换去发送其他任务请求），如果图片下载时间为2分钟，那么全部执行完毕也大概需要2分钟左右就可以了。

3. 异步编程

基于async & await关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术。

3.1 事件循环

事件循环，可以理解为一个while循环，这个while循环在周期性地运行并执行一些任务，在特定条件下终止循环。

# 伪代码

任务列表 = [任务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 函数名

协程对象，调用 协程函数() 得到的协程对象，比如：

# 定义一个协程函数
async def func():
    pass

# 调用协程函数，返回一个协程对象
result = func()

func就是一个协程函数，而result是协程对象

注意：调用协程函数，只是创建协程对象，函数内部代码并不会执行，只是会返回一个协程对象

如果想要运行协程函数内部代码，需要事件循环和协程对象配合才能实现，必须将协程对象交给事件循环来进行处理。

以下是一个简单的事件循环：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio 

async def func():
    print('哈哈哈')

# 调用协程函数，返回一个协程对象
result = func()

# 去生成一个事件循环
loop = asyncio.get_event_loop()

# 将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。
loop.run_until_complete(result)

注意：以上是早期的Python写法，如果是Python 3.7及以后的话，那么会有更简单的写法，如下：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/13 21:33
@Author  : boli.hong
"""
import asyncio 

async def func():
    print('哈哈哈')

result = func()

# 本质上方式一是一样的，内部先 创建事件循环 然后执行 run_until_complete，一个简便的写法。
asyncio.run(result)

这个过程可以简单理解为：将协程当做任务添加到事件循环的任务列表，然后事件循环检测列表中的协程是否已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。

3.3 await

await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起当前协程（任务），当前协程（任务）挂起过程中事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。

await + 可等待的对象（IO等待），包括协程对象、Future、Task对象

示例1：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio 

async def func():
    print('哈哈哈哈')
    
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await asyncio.sleep(3)
    
    print('结束', response)


asyncio.run(func())

示例2：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio 


async def others():
    print('开始')
    await asyncio.sleep(3)
    print('结束')
    return '返回值'


async def func():
    print('执行协程函数内部代码')

    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）
    response = await others()

    print('IO请求结束，结果为：', response)


asyncio.run(func())

运行：

执行协程函数内部代码
开始
结束
IO请求结束，结果为： 返回值

示例3：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio 


async def others():
    print('开始')
    await asyncio.sleep(3)
    print('结束')
    return '返回值'


async def func():
    print('执行协程函数内部代码')

    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）
    response1 = await others()
    print('IO请求结束，结果为：', response1)

    response2 = await others()
    print('IO请求结束，结果为：', response2)

asyncio.run(func())

await就是等待对象的值得到结果之后再继续往下走

上述的所有示例都只是创建了一个任务，即：事件循环的任务列表中只有一个任务，所以在IO等待时无法演示切换到其他任务效果。

在程序想要创建多个任务对象，需要使用Task对象来实现。

3.4 Task对象

Tasks are used to schedule coroutines concurrently.

When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon。

Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用asyncio.create_task()函数以外，还可以用低层级的loop.create_task()或ensure_future()函数，不建议手动实例化Task对象。

本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。

注意：asyncio.create_task()只适用于Python3.7及以上版本，低版本需要使用loop.create_task()或asyncio.ensure_future()

示例1：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio 


async def func():
    print('开始')
    await asyncio.sleep(3)
    print('结束')
    return '返回值'


async def main():
    print('main开始')

    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）
    task1 = asyncio.create_task(func())

    task2 = asyncio.create_task(func())

    print('main结束')

    # 当执行某协程遇到IO操作时，会自动化切换执行到其他任务
    # 此处的await是等待相应的协程全都执行完毕并获取结果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)

asyncio.run(main())

示例1的用法，用得比较少

示例2：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


async def func():
    print('开始')
    await asyncio.sleep(3)
    print('结束')
    return '返回值'


async def main():
    print('main开始')

    task_list = [
        asyncio.create_task(func()),
        asyncio.create_task(func()),
    ]

    print('main结束')

    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done
    # 如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。
    (done, pending) = await asyncio.wait(task_list)
    print(done)
    print(pending)

asyncio.run(main())

注意：Python3.8中的create_task函数新增了name参数，另外，await asyncio.wait返回的结果也和3.7不一样，主要是pending参数，Task字典里面会多了name的属性

注意：asyncio.wait 源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为[func(),func()] 也是可以的。

不设置timeout或者是timeout设置为大于3秒

main开始
main结束
开始
开始
结束
结束
{<Task finished coro=<func() done, defined at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:9> result='返回值'>, <Task finished coro=<func() done, defined at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:9> result='返回值'>}
set()

timeout设置为1-2秒

main开始
main结束
开始
开始
set()
{<Task pending coro=<func() running at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:11> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x039D5150>()]>>, <Task pending coro=<func() running at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:11> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x039D5130>()]>>}

没有打印出结束

timeout设置为3秒

main开始
main结束
开始
开始
set()
{<Task pending coro=<func() running at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:11> wait_for=<Future finished result=None>>, <Task pending coro=<func() running at D:/software/laragon/www/test/asyncio_tst/3.4_task2.py:11> wait_for=<Future finished result=None>>}
结束
结束

示例3：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


async def func():
    print('开始')
    await asyncio.sleep(3)
    print('结束')
    return '返回值'


task_list = [
    func(),
    func(),
]

# 错误：task_list = [ asyncio.create_task(func()), asyncio.create_task(func()) ]  
# 此处不能直接 asyncio.create_task，因为将Task立即加入到事件循环的任务列表，
# 但此时事件循环还未创建，所以会报错。

# 使用asyncio.wait将列表封装为一个协程，并调用asyncio.run实现执行两个协程
# asyncio.wait内部会对列表中的每个协程执行ensure_future，封装为Task对象
(done, pending) = asyncio.run(asyncio.wait(task_list))
print(done)
print(pending)

运行：

开始
开始
结束
结束
{<Task finished coro=<func() done, defined at D:/software/laragon/www/test/asyncio_tst/3.4_task3.py:9> result='返回值'>, <Task finished coro=<func() done, defined at D:/software/laragon/www/test/asyncio_tst/3.4_task3.py:9> result='返回值'>}
set()

3.5 asyncio.Future对象

A Futureis a special low-level awaitable object that represents an eventual result of an asynchronous operation.

asyncio中的Future对象是一个相对更偏向底层的可对象，通常我们不会直接用到这个对象，而是直接使用Task对象来完成任务的并和状态的追踪。（ Task 是 Futrue的子类）

Future为我们提供了异步编程中的最终结果的处理（Task类也具备状态处理的功能）。

示例1：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务（Future对象，如果没绑定任何行为，则这个任务永远不会结束）
    future = loop.create_future()
	
	# 等待Future对象获取最终结果，否则会一直等下去
	await future


asyncio.run(main())

示例2：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import asyncio


async def set_after(future):
    await asyncio.sleep(2)
    future.set_result("666")


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务（Future对象，如果没绑定任何行为，则这个任务永远不会结束）
    future = loop.create_future()

    # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2秒之后会给future赋值，即手动设置future任务的最终结果，那么future就可以结束了
    await loop.create_task(set_after(future))

    # 等待Future对象获取最终结果，否则会一直等下去
    data = await future
    
    print(data)


asyncio.run(main())

运行： 666

Future对象本身函数进行绑定，所以想要让事件循环获取Future的结果，则需要手动设置。而Task对象继承了Future对象，其实就对Future进行扩展，他可以实现在对应绑定的函数执行完成之后，自动执行set_result，从而实现自动结束。

虽然，平时使用的是Task对象，但对于结果的处理本质是基于Future对象来实现的。

扩展：支持 await 对象语法的对象课成为可等待对象，所以 协程对象、Task对象、Future对象 都可以被成为可等待对象。

3.6 futures.Future对象

在Python的concurrent.futures模块中也有一个Future对象，这个对象是基于线程池和进程池实现异步操作时使用的对象。

使用线程池、进程池来实现异步操作时用到的对象，即使用线程、进程做异步编程

两个Future对象是不同的，他们是为不同的应用场景而设计，例如：concurrent.futures.Future不支持await语法等。

官方提示两对象之间不同：

unlike asyncio Futures, concurrent.futures.Future instances cannot be awaited.
asyncio.Future.result() and asyncio.Future.exception() do not accept the timeout argument.
asyncio.Future.result() and asyncio.Future.exception() raise an InvalidStateError exception when the Future is not done.
Callbacks registered with asyncio.Future.add_done_callback() are not called immediately. They are scheduled with loop.call_soon() instead.
asyncio Future is not compatible with the concurrent.futures.wait() and concurrent.futures.as_completed() functions.

在Python提供了一个将futures.Future 对象包装成asyncio.Future对象的函数 asynic.wrap_future。

一般在程序开发中我们要么统一使用 asycio 的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果 协程的异步和 进程池/线程池的异步 混搭时，那么就会用到此功能了

注意：比起使用asyncio，使用线程、进程来做异步编程的话，耗费的资源会更高，这是因为协程是代码自己在单个线程中创建的，而进程、线程是操作系统本身支持的，切换进行/线程是需要进行操作系统级别的上下文切换，所需要的资源会更高，一般来说，优先使用协程，其次是线程，最后才是进程

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor


def func(value):
    print('func start：', time.strftime('%H:%M:%S', time.localtime(time.time())))
    time.sleep(3)
    print(value)
    print('func end：', time.strftime('%H:%M:%S', time.localtime(time.time())))
    print("----------\r\n")
    return value * 10


# 创建线程池，最多为5个线程
pool = ThreadPoolExecutor(max_workers=5)
# 或者是以下写法，创建进程池
# pool = ProcessPoolExecutor(max_workers=5)

for i in range(10):
    print('start：', time.strftime('%H:%M:%S', time.localtime(time.time())))
    # 使用线程池执行函数
    fut = pool.submit(func, i)
    print('result：', fut.result)
    print('end：', time.strftime('%H:%M:%S', time.localtime(time.time())))
    print("-----------------\r\n")

运行：

start： 23:04:48
func start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1acfa438 state=running>>
end： 23:04:48
-----------------

start： 23:04:48
func start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1add6940 state=running>>
end： 23:04:48
-----------------

start： 23:04:48
func start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1acfa4a8 state=running>>
end： 23:04:48
-----------------

start： 23:04:48
func start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa2e8 state=running>>
end： 23:04:48
-----------------

start： 23:04:48
func start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa4e0 state=running>>
end： 23:04:48
-----------------

start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa6a0 state=pending>>
end： 23:04:48
-----------------

start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa748 state=pending>>
end： 23:04:48
-----------------

start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa828 state=pending>>
end： 23:04:48
-----------------

start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa908 state=pending>>
end： 23:04:48
-----------------

start： 23:04:48
result： <bound method Future.result of <Future at 0x1db1aeaa9e8 state=pending>>
end： 23:04:48
-----------------

0
func end： 23:04:51
----------

func start： 23:04:51
1
func end： 2
func end： 23:04:51
----------

23:04:51
----------
func start： 23:04:51

func start： 23:04:51
3
func end： 23:04:51
----------

func start： 23:04:51
4
func end： 23:04:51
----------

func start： 23:04:51
5
func end： 23:04:54
----------

7
func end： 68
func end： 23:04:54
23:04:54
----------

9
func end： 23:04:54
func end： 23:04:54
----------
----------



----------

可以看出：

使用线程池来执行10次func函数，23:04:48在time.sleep之前就全部打印出来了，Future对象的result的状态为：pending
sleep之后，分为两个时间进行打印，分别是23:04:51和23:04:54，这是因为我们只开了5个线程，而有10个任务，所以需要分为两次执行
注意：sleep之后，打印的结果的顺序并不是绝对的，比如func end：68这一行，是因为有两个进程同时返回了结果，同时打印在了控制台，导致打印的结果是乱的

该对象的使用场景：

在异步编程中，有一些组件或第三方模块不支持协程，比如说requests、MySQL等，不支持协程，所以遇到这种情况需要使用线程、进程来做异步编程

比如：CRM项目中80%都是基于协程异步编程+MySQL

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
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())

运行：

default thread pool SB

代码中的fut = loop.run_in_executor(None, func1)就相当于：

pool = ThreadPoolExecutor()
fut = pool.submit(func1)

由于返回的是concurrent.futures.Future对象，该对象不支持await语法，所以需要包装为asycio.Future对象，即result = await fut

另一种写法为：

loop = asyncio.get_running_loop()

with concurrent.futures.ThreadPoolExecutor() as pool:
    result = await loop.run_in_executor(
        pool, func1)
    print('custom thread pool', result)

另外，也可以使用进程：

loop = asyncio.get_running_loop()

with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, func1)
        print('custom process pool', result)

案例：asyncio + 不支持异步的模块，比如requests

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/6/25
@Author  : hongboli
"""
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://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg',
        'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg',
        'https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg'
    ]

    tasks = [download_image(url) for url in url_list]

    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

运行：

开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918536758&di=5a9d49f622a3c2f092830713174a89d2&imgtype=0&src=http%3A%2F%2Fimg.pconline.com.cn%2Fimages%2Fupload%2Fupc%2Ftx%2Fwallpaper%2F1305%2F16%2Fc4%2F20990657_1368686545119.jpg
开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918729593&di=52738a7ab704a17c742c89a7f1b79ec3&imgtype=0&src=http%3A%2F%2Fe.hiphotos.baidu.com%2Fzhidao%2Fpic%2Fitem%2F7dd98d1001e939014d91b6e079ec54e737d19658.jpg
开始下载: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1590918597135&di=69ded1b30923b75e6baa210889bdbeef&imgtype=0&src=http%3A%2F%2Fpic1.win4000.com%2Fwallpaper%2F2017-12-29%2F5a45a53b2763d.jpg
下载完成
下载完成
下载完成

3.7 异步迭代器

什么是异步迭代器

实现了 __aiter__() 和 __anext__() 方法的对象。__anext__ 必须返回一个 awaitable 对象。async for 会处理异步迭代器的 __anext__() 方法所返回的可等待对象，直到其引发一个 StopAsyncIteration 异常。由 PEP 492 引入。

什么是异步可迭代对象？

可在 async for 语句中被使用的对象。必须通过它的 __aiter__() 方法返回一个 asynchronous iterator。由 PEP 492 引入。

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语法而已。

3.8 异步上下文管理

此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。

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

只要对象里面定义了__aenter__，就可以使用async with 对象实例 as f，当然，同样和上个例子一样，async with同样是需要放在async def函数内的

这个异步的上下文管理器还是比较有用的，平时在开发过程中打开、处理、关闭操作时，就可以用这种方式来处理。

3.9 小结

在程序中只要看到async和await关键字，其内部就是基于协程实现的异步编程，这种异步编程是通过一个线程在IO等待时间去执行其他任务，从而实现并发。

以上就是异步编程的常见操作，内容参考官方文档。

中文版：https://docs.python.org/zh-cn/3.8/library/asyncio.html
英文本：https://docs.python.org/3.8/library/asyncio.html

4. uvloop

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

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

安装uvloop

pip3 install uvloop

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

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

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

# 内部的事件循环自动化会变为uvloop
asyncio.run(...)

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

5.实战案例

为了更好理解，上述所有示例的IO情况都是以 asyncio.sleep 为例，而真实的项目开发中会用到很多IO的情况。

5.1 异步redis

当通过python去操作redis时，链接、设置值、获取值这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

在使用python代码操作redis时，链接/操作/断开都是网络IO。

示例1：异步操作redis：

安装Python异步操作redis模块

pip3 install aioredis

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
import asyncio
import aioredis


async def execute(address, password=None):
    print("开始执行", address)

    # 网络IO操作：创建redis连接
    if password:
        redis = await aioredis.create_redis(address, password=password)
    else:
        redis = await aioredis.create_redis(address)

    # 网络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://127.0.0.1:6379'))

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

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
import asyncio
import aioredis


async def execute(address, password=None):
    print("开始执行", address)

    # 网络IO操作：先去连接第一个 ，遇到IO则自动切换任务，去连接另一个
    if password:
        redis = await aioredis.create_redis_pool(address, password=password)
    else:
        redis = await aioredis.create_redis_pool(address)

    # 网络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://127.0.0.1:6379'),
    execute('redis://127.0.0.1:6379')
]

asyncio.run(asyncio.wait(task_list))

更多redis操作参考aioredis官网：https://aioredis.readthedocs.io/en/v1.3.0/start.html

5.2 异步MySQL

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

安装Python异步操作redis模块

pip3 install aiomysql

示例1：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
import asyncio
import aiomysql


async def execute():
    # 网络IO操作：连接MySQL
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123456', 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：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
import asyncio
import aiomysql


async def execute(host, password):
    print("开始", host)
    # 网络IO操作：先去连接 127.0.0.1，遇到IO则自动切换任务，去连接另一个127.0.0.1
    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('127.0.0.1', "123456"),
    execute('127.0.0.1', "123456")
]

asyncio.run(asyncio.wait(task_list))

5.3 FastApi框架

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

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

安装FastAPI web 框架，

pip3 install fastapi

安装uvicorn，本质上为web提供socket server的支持的asgi（一般支持异步称asgi、不支持异步称wsgi）

pip3 install uvicorn # 内部是基于uvloop

示例：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI

app = FastAPI()

# 创建redis的连接池
REDIS_POOL = aioredis.ConnectionsPool('redis://127.0.0.1:6379', 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("5.3_fastapi:app", host="127.0.0.1", port=5000, log_level="info")

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

例如：同时有两个用户并发来向接口 http://127.0.0.1:5000/red 发送请求，服务端只有一个线程，同一时刻只有一个请求被处理。

异步处理可以提供并发是因为：当视图函数在处理第一个请求时，第二个请求此时是等待被处理的状态，当第一个请求遇到IO等待时，会自动切换去接收并处理第二个请求，当遇到IO时自动化切换至其他请求，一旦有请求IO执行完毕，则会再次回到指定请求向下继续执行其功能代码。

5.4 爬虫

在编写爬虫应用时，需要通过网络IO去请求目标数据，这种情况适合使用异步编程来提升性能，接下来我们使用支持异步编程的aiohttp模块来实现。

安装aiohttp模块

pip3 install aiohttp

示例：

# -*- encoding: utf-8 -*-
"""
@Time    : 2020/7/12
@Author  : hongboli
"""
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))


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)
        print(done)
        print(pending)


if __name__ == '__main__':
    asyncio.run(main())

总结

最大的意义：通过一个线程利用其IO等待时间去做一些其他事情。

为了提升性能越来越多的框架都在向异步编程靠拢，例如：sanic、tornado、django3.0、django channels组件等，用更少资源可以做处理更多的事，何乐而不为呢。

Paul / asyncio_demo

协程 & asyncio & 异步编程

1. 协程

1.1 greenlet实现协程

1.2 yield关键字

1.3 asyncio

1.4 async & await 关键字

2. 协程的意义

2.1 爬虫案例

3. 异步编程

3.1 事件循环

3.2 快速上手

3.3 await

3.4 Task对象

3.5 asyncio.Future对象

3.6 futures.Future对象

3.7 异步迭代器

3.8 异步上下文管理

3.9 小结

4. uvloop

5.实战案例

5.1 异步redis

5.2 异步MySQL

5.3 FastApi框架

5.4 爬虫

总结

简介

发行版

贡献者

近期动态

Paul / asyncio_demo .gitee-modal { width: 500px !important; }

协程 & asyncio & 异步编程

1. 协程

1.1 greenlet实现协程

1.2 yield关键字

1.3 asyncio

1.4 async & await 关键字

2. 协程的意义

2.1 爬虫案例

3. 异步编程

3.1 事件循环

3.2 快速上手

3.3 await

3.4 Task对象

3.5 asyncio.Future对象

3.6 futures.Future对象

3.7 异步迭代器

3.8 异步上下文管理

3.9 小结

4. uvloop

5.实战案例

5.1 异步redis

5.2 异步MySQL

5.3 FastApi框架

5.4 爬虫

总结

简介

发行版

贡献者

近期动态

搜索帮助

Paul / asyncio_demo