Python Asyncio 所有异步协程库用法详解_python协程await后哪些类型

---

title: Asyncio并发编程
copyright: true
top: 0
date: 2019-04-03 14:09:24
tags: Asyncio
categories: Python高阶笔记
permalink:
password:
keywords:
description: Python高并发框架，进一步解读asyncio的语法与功能，同时对支持异步的web网络请求包aiohttp功能做介绍。

我喜欢铁路，你沿着铁路走，在尽头肯定会找到一座城市，或者其他什么有人的地方。不像鸟飞在空中，甚至不知道前面会不会有目的地。

在此之后可以尝试阅读Python 异步协程概念，协程是一个很大的框架，需要从基础原理慢慢学习。

Asyncio

在python3.5之前，都是使用生成器的一些技巧完成协程任务，他们的调度方式依然是 事件循环+协程模式。这样设计结构和维护虽然相对于回调函数简单一些，但是代码还是有一些混乱，并且又当作生成器又当作协程，都是还是一些技巧性的东西，为了将语义变得更加明确，于是在python3.5使用了async和await(功能与yield from类似)关键词正式定义原生协程，asyncio是python解决异步io编程的一个完整框架。

它具有如下定义：

1. 包含各种特定系统实现的模块化事件循环
2. 传输与协议抽象
3. 对TCP,UDP,SSL,子进程，延时调用以及其他的具体支持
4. 模仿futures模块适用于事件循环使用到Future类
5. 基于yield from的协议和任务，可以使用顺序执行的方式编写并发代码
6. 必须使用一个将产生阻塞IO的调用时，有接口可以把这个事件转移到线程池
7. 模仿threading模块中的同步语法，可以用在单线程内实现协程同步

协程编程离不开的三大要点：

1. 事件循环
2. 回调(驱动生成器)
3. epoll/select(IO多路复用)

Asyncio是一个异步编程的框架，可以解决异步编程，协程调度问题，线程问题，是整个异步IO的解决方案。

事件循环

简单案例(访问一个网站)

async def get_url_title(url):
# 使用关键词async定义一个协程
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    # 这一步至关重要
    # asyncio.sleep(2) 功能:异步非阻塞等待2s，作用是模拟访问网站消耗的时间
    # await 的作用类似 yield，即这个时候把线程资源控制权交出去,监听这个描述符直到这个任务完成
    # await 后面只能接三种类型
    '''
    1. 协程:Python 协程属于 可等待 对象，因此可以在其他协程中被等待:
    2. 任务:任务 被用来设置日程以便 并发 执行协程。(当一个协程通过 asyncio.create_task() 等函数被打包为一个 任务，该协程将自动排入日程准备立即运行)
    3. Future 对象:Future 是一种特殊的 低层级 可等待对象，表示一个异步操作的 最终结果。(当一个 Future 对象 被等待，这意味着协程将保持等待直到该 Future 对象在其他地方操作完毕。)
    
    如果await time.sleep(2) 是会报错的
    '''
    print('网站访问成功')

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # 一行代码创造事件循环
    loop.run_until_complete(get_url_title('http://www.langzi.fun'))
    # 这是一个阻塞的方法,可以理解成多线程中的join方法
    # 直到get_url_title('http://www.langzi.fun')完成后，才会继续执行下面的代码
    end_time = time.time()
    print('消耗时间:{}'.format(end_time-start_time))
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

返回结果：

开始访问网站:http://www.langzi.fun
网站访问成功
消耗时间:2.0018768310546875
1
2
3

简单案例(访问多个网站)

协程的优势是多任务协作，单任务访问网站没法发挥出他的功能，一次性访问多个网站或者一次性等待多个IO响应时间才能发挥它的优势。

# -*- coding:utf-8 -*-
import asyncio
import time

async def get_url_title(url):
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    print('网站访问成功')

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # 创造一个事件循环
    tasks = [get_url_title('http://www.langzi.fun')for i in range(10)]
    # 这个列表代表总任务量，即执行10次get_url_title()函数
    loop.run_until_complete(asyncio.wait(tasks))
    # asyncio.wait后面接上非空可迭代对象,一般来说是功能函数列表
	# 功能是一次性提交多个任务，等待完成
	# loop.run_until_complete(asyncio.gather(*tasks))
	# 和上面代码功能一致，但是gather更加高级，如果是列表就需要加上*
    # 这里会等到全部的任务执行完后才会执行后面的代码
    end_time = time.time()
    print('消耗时间:{}'.format(end_time-start_time))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

对一个网站发起10次请求，返回结果：

开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
消耗时间:2.0015649795532227
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

gather与wait的区别：

• gather更擅长于将函数聚合在一起
• wait更擅长筛选运行状况

即gather更加高级，他可以将任务分组，也可以取消任务

import asyncio

async def get_url_title(url):
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    print('网站访问成功')
    return 'success'

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # 使用wait方法
    # tasks = [get_url_title('http://www.langzi.fun')for i in range(10)]
    # loop.run_until_complete(asyncio.wait(tasks))

    # 使用gather方法实现分组导入(方法1)
    group1 = [get_url_title('http://www.langzi.fun')for i in range(3)]
    group2 = [get_url_title('http://www.baidu.com')for i in range(5)]
    loop.run_until_complete(asyncio.gather(*group1,*group2))
    # 这种方法会把两个全部一次性导入

    # 使用gather方法实现分组导入(方法2)
    group1 = [get_url_title('http://www.langzi.fun')for i in range(3)]
    group2 = [get_url_title('http://www.baidu.com')for i in range(5)]
    group1 = asyncio.gather(*group1)
    group2 = asyncio.gather(*group2)
    #group2.cancel() 取消group2任务
    loop.run_until_complete(asyncio.gather(group1,group2))
    # 这种方法会先把group1导入，然后导入group2
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

返回结果：

开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
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

另外一种使用gather获取返回结果:

import asyncio

async def get_url_title(url):
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    print('网站访问成功')
    return 'success'

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # 使用gather方法传递任务获取结果
    group1 = asyncio.ensure_future(get_url_title('http://www.langzi.fun'))
    loop.run_until_complete(asyncio.gather(group1))
	# 如果不是列表就不需要加*
    print(group1.result())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

返回结果：

开始访问网站:http://www.langzi.fun
网站访问成功
success
1
2
3

还有一些复杂的区别转移到python 异步协程中查看

协程的调用和组合十分灵活，尤其是对于结果的处理，如何返回，如何挂起，需要逐渐积累经验和前瞻的设计。

简单案例(获取返回值)

# -*- coding:utf-8 -*-
import asyncio
import time

async def get_url_title(url):
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    print('网站访问成功')
    return 'success'

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # 创建一个事件循环

    get_future = loop.create_task(get_url_title('http://www.langzi.fun'))
    #get_future = asyncio.ensure_future(get_url_title('http://www.langzi.fun'))
    # 这两行代码功能用法一模一样

    loop.run_until_complete(get_future)
    print('获取结果:{}'.format(get_future.result()))
    # 获取结果
    
    end_time = time.time()
    print('消耗时间:{}'.format(end_time-start_time))
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

返回结果：

开始访问网站:http://www.langzi.fun
网站访问成功
获取结果:success
消耗时间:2.0019724369049072
1
2
3
4

如果是多个网址传入，访问多个网址的返回值呢？只需要把前面的知识点汇总一起即可使用：

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # 创建一个事件循环

    tasks = [loop.create_task(get_url_title('http://www.langzi.fun')) for i in range(10)]
	# 把所有要返回的函数加载到一个列表

    loop.run_until_complete(asyncio.wait(tasks))
	# 这里和上面用法一样

    print('获取结果:{}'.format([x.result() for x in tasks]))
    # 因为结果都在一个列表，在列表中取值即可

    end_time = time.time()
    print('消耗时间:{}'.format(end_time-start_time))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

返回结果：

开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
获取结果:['success', 'success', 'success', 'success', 'success', 'success', 'success', 'success', 'success', 'success']
消耗时间:2.0016491413116455
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

简单案例(回调函数)

上面的例子是一个协程函数，当这个协程函数的await xxx执行完毕后，想要执行另一个函数后，然后再返回这个协程函数的返回结果该这么做：

# -*- coding:utf-8 -*-
import asyncio
from functools import partial
# partial的功能是 固定函数参数，返回一个新的函数。你可以这么理解：
'''
from functools import partial
    def go(x,y):
        return x+y
    g = partial(go,y=2)
    print(g(1))
返回结果：3

    g = partial(go,x=5,y=2)
    print(g())
返回结果：7
    
'''
async def get_url_title(url):
    print('开始访问网站:{}'.format(url))
    await asyncio.sleep(2)
    print('网站访问成功')
    # 当这个协程函数快要结束返回值的时候，会调用下面的call_back函数
    # 等待call_back函数执行完毕后，才返回这个协程函数的值
    return 'success'

def call_back(future,url):
    # 注意这里必须要传递future参数，因为这里的future即代表下面的get_future对象
    print('检测网址:{}状态正常'.format(url))

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # 创建一个事件循环

    get_future = loop.create_task(get_url_title('http://www.langzi.fun'))
    # 将一个任务注册到loop事件循环中

    get_future.add_done_callback(partial(call_back,url = 'http://www.langzi.fun'))
    # 这里是设置，当上面的任务完成要返回结果的时候，执行call_back函数
    # 注意call_back函数不能加上()，也就意味着你只能依靠partial方法进行传递参数
    
    loop.run_until_complete(get_future)
    # 等待任务完成
    print('获取结果:{}'.format(get_future.result()))
    # 获取结果
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
39
40
41
42
43
44

返回结果：

开始访问网站:http://www.langzi.fun
网站访问成功
检测网址:http://www.langzi.fun状态正常
获取结果:success
1
2
3
4

梳理

1. 协程函数必须要使用关键词async定义
2. 如果遇到了要等待的对象，必须要使用await
3. 使用await后面的任务，必须是可等待对象(三种主要类型: 协程, 任务 和 Future.)
4. 运行前，必须要创建一个事件循环(loop = asyncio.get_event_loop(),一行代码即可)
5. 然后把任务加载到该事件循环中即可
6. 如果需要获取协程函数的返回值，需要使用loop.create_task()或asyncio.ensure_future()函数，在最后使用.result()获取返回结果。
7. 如果想要把多个任务注册到loop中，需要使用一个列表包含他们，调用的时候使用asyncio.wait(list)

取消协程任务

存在多个任务协程，想使用ctrl c退出协程，使用例子讲解：

import asyncio
async def get_time_sleep(t):
    print('开始运行，等待:{}s'.format(t))
    await asyncio.sleep(t)
    print('运行结束')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # 创建一个事件循环
    task_1 = get_time_sleep(1)
    task_2 = get_time_sleep(2)
    task_3 = get_time_sleep(3)

    tasks = [task_1,task_2,task_3]
    # 三个协程任务加载到一个列表

    try:
        loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt:
        # 当检测到键盘输入 ctrl c的时候
        all_tasks = asyncio.Task.all_tasks()
        # 获取注册到loop下的所有task
        for task in all_tasks:
            print('开始取消协程')
            task.cancel()
            # 取消该协程,如果取消成功则返回True
        loop.stop()
		# 停止循环
        loop.run_forever()
		# loop事件循环一直运行
        # 这两步必须要做
    finally:
        loop.close()
		# 关闭事件循环
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

run_forever 会一直运行，直到 stop 被调用，但是你不能像下面这样调 stop

loop.run_forever()
loop.stop()
1
2

run_forever 不返回，stop 永远也不会被调用。所以，只能在协程中调 stop：

async def do_some_work(loop, x):
    print('Waiting ' + str(x))
    await asyncio.sleep(x)
    print('Done')
    loop.stop()
1
2
3
4
5

这样并非没有问题，假如有多个协程在 loop 里运行：

asyncio.ensure_future(do_some_work(loop, 1))
asyncio.ensure_future(do_some_work(loop, 3))

loop.run_forever()
1
2
3
4

第二个协程没结束，loop 就停止了——被先结束的那个协程给停掉的。
要解决这个问题，可以用 gather 把多个协程合并成一个 future，并添加回调，然后在回调里再去停止 loop。

async def do_some_work(loop, x):
    print('Waiting ' + str(x))
    await asyncio.sleep(x)
    print('Done')

def done_callback(loop, futu):
    loop.stop()

loop = asyncio.get_event_loop()

futus = asyncio.gather(do_some_work(loop, 1), do_some_work
1
2
3
4
5
6
7
8
9
10