Python 协程模块 asyncio 使用指南

前面我们通过5 分钟入门 Python 协程了解了什么是协程,协程的优点和缺点和如何在 Python 中实现一个协程。没有看过的同学建议去看看。这篇文章，将不再对理论性的东西做过多的解说。而是倾向于 asyncio 的使用上，另外为了保证文章时效性这里我们使用 Python3.8 来进行对后面内容的操作。

协程的演变

其实早在 Python3.4 的时候就有协程,当时的协程是通过 @asyncio.coroutine 和 yeild from 实现的。在一些很老教程中你可能看到的是下面这种形式:

import asyncio

@asyncio.coroutine

def print_hello():

print("Hello world!")

r = yield from asyncio.sleep(1)

print("Hello again!")

# 创建并获取EventLoop:

loop = asyncio.get_event_loop()

# 执行协程

loop.run_until_complete(print_hello())

loop.close()

因为现在几乎没有人这样写了所以仅作为了解即可。

然后到了 Python3.5 引入了 async/await 语法糖，一直到现在Python3.8 都是用这种形式来表示协程,示例如下。

import asyncio

async def print_hello():

print("Hello world!")

await asyncio.sleep(1)

print("Hello again!")

if __name__ == '__main__':

loop = asyncio.get_event_loop()

try:

print("开始运行协程")

coro = print_hello()

print("进入事件循环")

loop.run_until_complete(coro)

finally:

print("关闭事件循环")

loop.close()

这种是目前应用范围最广的，可以看到比之前的代码舒服了不少，不用再使用装饰器的形式了。然后就到了 Python3.7 和 Python3.8 ,协程发生了很多细小的变化，但是最大的一个变化就是，启动协程的方法变简单了，一句就可以搞定，不用再像上面那样创建循环然后再仍到事件循环去执行了。

import asyncio

async def print_hello():

print("Hello world!")

await asyncio.sleep(1)

print("Hello again!")

if __name__ == '__main__':

print("开始运行协程")

asyncio.run(print_hello())

print("进入事件循环")

怎么样是不是代码更少了，启动协程更简单了。所以这也正是我们使用 3.8 作为本教程的 Python 版，与时俱进嘛。

Asyncio 的组成部分

根据目前的官方文档，总的来说分为了两部分:高层级 API 和低层级API。

首先看高层级 API 也是接下来重点要讲的。

高层级API

协程对象和 Tasks 对象

数据流

同步源语

子进程

队列

异常

低层级API

事件循环

Futures 对象

传输和协议

策略

平台支持

上面列出了这么多的项目我们怎么去选择自己所需要的呢，总的来说对于刚入门的新手或者只是写一个自己用的程序一般都只会用到高级 API 的部分，这部分就属于开箱即用的那种，对于高级用户比如框架开发者，往往可以需要去适应各种需要，需要重新改写一些内部的结构，这个时候就需要用到低层级的 API，但是这两个层级呢只能是一个大概方向吧，主要是方便 API 的查看，下面呢我将围绕者高层级API和低层级API在日常实际工作中经常用到的内容做一些讲解。

了解几个概念

在学习 asyncio 之前需要知道这样的几个概念。

事件循环

事件循环是一种处理多并发量的有效方式，在维基百科中它被描述为「一种等待程序分配事件或消息的编程架构」，我们可以定义事件循环来简化使用轮询方法来监控事件，通俗的说法就是「当A发生时，执行B」。所谓的事件，其实就是函数。事件循环，就是有一个队列，里面存放着一堆函数，从第一个函数开始执行，在函数执行的过程中，可能会有新的函数继续加入到这个队列中。一直到队列中所有的函数被执行完毕，并且再也不会有新的函数被添加到这个队列中,程序就结束了。

Future

Future 是一个数据结构，表示还未完成的工作结果。事件循环可以监视Future 对象是否完成。从而允许应用的一部分等待另一部分完成一些工作。

简答说，Future 就是一个类，用生成器实现了回调。

Task

Task 是 Future 的一个子类，它知道如何包装和管理一个协程的执行。任务所需的资源可用时，事件循环会调度任务允许，并生成一个结果，从而可以由其他协程消费。一般操作最多的还是 Task。用Task来封装协程，给原本没有状态的协程增加一些状态。

awaitable objects(可等待对象)

如果一个对象可以用在 wait 表达式中，那么它就是一个可等待的对象。在 asyncio 模块中会一直提到这个概念，其中协程函数，Task，Future 都是 awaitable 对象。

用于 await 表达式中的对象。可以是的 coroutine 也可以是实现了 _await_() 方法的对象，参见 PEP 492。类比于 Iterable 对象是 Generator 或实现了_iter_() 方法的对象。

object._await_(self)

必须返回生成器，asyncio.Future 类也实现了该方法，用于兼容 await 表达式。

而 Task 继承自 Future，因此 awaitable 对象有三种：coroutines、Tasks 和 Futures。

await 的目的：

获取协程的结果

挂起当前协程，将控制交由事件循环，切换到其他协程，然后等待结果，最后恢复协程继续执行

启动一个协程

现在我们使用 async/await 语法来声明一个协程。 代码如下

import asyncio

async def main():

print('hello')

await asyncio.sleep(1)

print('world')

if __name__ == '__main__':

asyncio.run(main())

asyncio.run 只能用来启动程序入口协程,反过来你在程序中如果使用asyncio.run 就会出错，直接我们提到对于其他的协程通过await链来实现，这里也是一样的。下面说下代码的含义，首先启动 main 这个协程，main 方法就是先打印 hello，然后在打印过程中通过使用 asyncio.sleep 来等待1秒,之后再打印 world。前面我们提到用协程就意味着我们要一直使用非阻塞的代码，才能达到速度提升，所以这里我们用了非阻塞版的 time.sleep 即 asyncio.sleep 。

协程中调用协程

之前我们提到了在协程中，可以使用 await 来调用一个协程。

就像下面的代码:

import asyncio

import time

async def say_after(delay, what):

await asyncio.sleep(delay)

print(what)

async def main():

#使用f-string拼接字符串

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

child1=await say_after(1, 'hello') #通过await调用协程，然后接收一下返回值

child2=await say_after(2, 'world')

print("child1",child1)

print("child2",child2)

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

if __name__ == '__main__':

asyncio.run(main())

运行结果：

开始运行 11:17:26

hello

world

child1 None

child2 None

结束运行 11:17:29

[Finished in 3.1s]

代码是没什么问题,正常运行。但是一般情况下我们用到更多的是下面的方式。将协程封装为 Task 让原本没有状态标示的协程添加上状态 。

我们可以通过 asyncio.create_task 方法来实现。

asyncio.create_task

create_task(在3.6版本中需要使用低层级的API asyncio.ensure_future。)是 3.7以后加入的语法,作用是将协程包装为一个任务(Task),相比3.6版本的ensure_future可读性提高。

将上面的代码做如下修改。

import asyncio

import time

async def say_after(delay, what):

await asyncio.sleep(delay)

print(what)

async def main():

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

child1=asyncio.create_task(say_after(1, 'hello')) #通过await调用协程，然后接收一下返回值

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

print("调用任务child1前",child1)

print("调用任务child2前",child2)

await child1

await child2

print("调用任务child1后",child1)

print("调用任务child2前",child2)

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

if __name__ == '__main__':

asyncio.run(main())

运行结果如下:

开始运行 11:37:54

调用任务child1前 >

调用任务child2前 >

hello

world

调用任务child1后 result=None>

调用任务child2前 result=None>

结束运行 11:37:56

可以发现,我们的结果中多了"

前面说到 Task 是 Future 的子类，所以 Tas k拥有 Future 的一些状态。

Future的状态

大概有如下几种:

Pending

Running

Done

Cancelled

创建 future 的时候，task 为 pending，事件循环调用执行的时候当然就是 running，调用完毕自然就是 done，如果需要停止事件循环，就需要先把 task 取消，状态为 cancel。这里先做了解知道 Task 是有状态的就够了。

并发运行任务

一系列的协程可以通过 await 链式的调用，但是有的时候我们需要在一个协程里等待多个协程，比如我们在一个协程里等待 1000 个异步网络请求，对于访问次序有没有要求的时候，就可以使用另外的关键字asyncio.wait 或 asyncio.gather 来解决了。

asyncio.gather

使用方法

asyncio.gather(*aws, loop=None, return_exceptions=False)¶

也就是说使用 gather 语句并发协程，就得用 await 去执行它。这个方法可以接收三个参数,第一个 aws,

aws 一般是一个列表，如果里面的元素是 awaitable 类型，在运行的时候它将自动被包装成 Task，gather 会根据 aws 中元素添加的顺序。顺序执行并返回结果列表。

第二个 loop 可以传入一个事件循环对象，一般不用管,最后一个return_exceptions 默认是 False，如果 return_exceptions 为 True，异常将被视为成功结果，然后添加到结果列表中。

下面是一个10个数字并输出例子。

import asyncio

async def foo(num):

return num

async def main():

coro = [asyncio.create_task(foo(i)) for i in range(10) ]

done= await asyncio.gather(*coro)

for i in done:

print(i)

if __name__ == '__main__':

asyncio.run(main())

运行之后结果如下

0

1

2

3

4

5

6

7

8

9

gather 返回的结果是一个列表，迭代这个列表可以看到任务依次输出。

gather 通常被用来阶段性的一个操作，做完第一步才能做第二步，比如下面这样

import asyncio

import time

async def step1(n, start):

await asyncio.sleep(n)

print("第一阶段完成")

print("此时用时", time.time() - start)

return n

async def step2(n, start):

await asyncio.sleep(n)

print("第二阶段完成")

print("此时用时", time.time() - start)

return n

async def main():

now = time.time()

result = await asyncio.gather(step1(5, now), step2(2, now))

for i in result:

print(i)

print("总用时", time.time() - now)

if __name__ == '__main__':

asyncio.run(main())

输出内容

第二阶段完成

此时用时 2.0041821002960205

第一阶段完成

此时用时 5.0009942054748535

5

2

总用时 5.001508951187134

可以通过上面结果得到如下结论：

1.step1 和 step2 是并行运行的。

2.gather 会等待最耗时的那个完成之后才返回结果，耗时总时间取决于其中任务最长时间的那个。

asyncio.task

我们先看一下 wait 的语法结构:

asyncio.wait(aws, *, loop=None, timeout=None, return_when=ALL_COMPLETED)¶

wait 一共有 4 个参数,第一个参数 aws，一般是一个任务列表。

第二个*之后的都是强制关键字参数，即loop,timeout,return_when。

loop通gather的参数是一个事件循环,该参数计划在Python 3.10中删除。

timeout可以指定这组任务的超时时间，请注意，此函数不会引发asyncio.TimeoutErro, 超时的时候会返回已完成的任务。

return_when可以指定什么条件下返回结果，默认是所以任务完成就返回结果列表。return_when的具体参数看下面的表格:

参数名

含义

FIRST_COMPLETED

任何一个future完成或取消时返回

FIRST_EXCEPTION

任何一个future出现错误将返回，如果出现异常等价于ALL_COMPLETED

ALL_COMPLETED

当所有任务完成或者被取消时返回结果，默认值。

wait返回的结果是一个元组,第一部分是完成的任务,第二部分是准备中的任务。

done, pending = await asyncio.wait(aws)

其中done表示是完成的任务，可以通过迭代获取每个任务。

pending表示的是还没执行的任务。

下面看一个例子来进一步了解

import asyncio

async def foo(num):

await asyncio.sleep(0.99991)

return num

async def main():

#coro = foo()

coro = [asyncio.create_task(foo(i)) for i in range(10) ]

done, pending = await asyncio.wait(coro,timeout=1,return_when="ALL_COMPLETED")

for coro in done:

print(coro.result())

print("pending",pending)

for item in pending:

print(item)

if __name__ == '__main__':

asyncio.run(main())

运行结果如下:

2

5

3

0

6

4

1

7

pending { wait_for=>, wait_for=>}

wait_for=>

wait_for=>

首先说代码,使用wait实现并发的程序是无序的所以我们看到数字不是一次出现的。这个是和gather的不同之处,另外在返回的参数上也有差别,wait返回两个参数done和pending。

上面的代码指定了一个timeout，因为任务没在指定时间完成所以就导致，只有完成的任务输出了结果，没有完成的部分可以看到它们的状态是pending。

总结

最后我们,总结一下wait和gather的相同之处和不同之处:

相同之处:都可以完成多个任务的并发操作。

不同之外:gather适合按照顺序去做的任务,或者按照阶段去做的任务，返回的是结果列表，而wait不讲究任务的顺序，这个在做爬虫中经常使用到，然后wait可以返回2个结果，done和pending。

任务完成时处理

asyncio.as_completed

as_complete是一个生成器，会管理指定的一个任务列表，并生成他们的结果。每个协程结束运行时一次生成一个结果。与wait一样，as_complete不能保证顺序，不过执行其他动作之前没有必要等待所以后台操作完成。

我们看下这个函数都有哪些参数

asyncio.as_completed(aws, *, loop=None, timeout=None)

和前面的wait类似，第一个参数awas,然后loop，最后timeout,需要注意的是timeout如果指定了，那么在指定时间没完成的话会抛出asyncio.exceptions.TimeoutError异常。

下面看一个例子:

import asyncio

import time

async def foo(n):

print(f'等待{n}秒')

await asyncio.sleep(n)

return n

async def main():

coroutine1 = foo(1)

coroutine2 = foo(2)

coroutine3 = foo(4)

tasks = [asyncio.create_task(coroutine1),asyncio.create_task(coroutine2),asyncio.create_task(coroutine3)]

for task in asyncio.as_completed(tasks):

result = await task

print(f'获取返回结果: {result}')

if __name__ == '__main__':

now = lambda : time.time()

start = now()

asyncio.run(main())

print(now() - start)

输出结果

等待1秒

等待2秒

等待4秒

获取返回结果: 1

获取返回结果: 2

获取返回结果: 4

4.002715826034546

可以看出整个执行过程总用时取决于 等待时间最长的那个即4秒。

接下来，对上面的代码稍作修改，

将

for task in asyncio.as_completed(tasks):

改为

for task in asyncio.as_completed(tasks,timeout=2):

其他地方不变,改完运行之后会看到上面提到的错误。

等待1秒

等待2秒

等待4秒

获取返回结果: 1

Traceback (most recent call last):

File "/Users/chennan/Desktop/2019/aiochatuse/ascomplete.py", line 25, in

asyncio.run(main())

File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/runners.py", line 43, in run

return loop.run_until_complete(main)

File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 589, in run_until_complete

return future.result()

File "/Users/chennan/Desktop/2019/aiochatuse/ascomplete.py", line 18, in main

result = await task

File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/tasks.py", line 570, in _wait_for_one

raise exceptions.TimeoutError

asyncio.exceptions.TimeoutError

取消任务的时候保证其他协程运行完毕

在取消任务的时候存在一个问题,首先先看一段代码:

import asyncio

async def coro():

print('开始休眠')

await asyncio.sleep(2)

print('结束休眠')

async def cancel_it(some_task):

await asyncio.sleep(0.5)

some_task.cancel()

print('取消some_task任务')

async def main():

real_task = asyncio.create_task(coro())

await cancel_it(real_task)

await real_task

if __name__ == '__main__':

asyncio.run(main())

运行之后你会看到如下结果

开始休眠

取消some_task任务

Traceback (most recent call last):

File "/Users/chennan/Desktop/2019/aiochatuse/shielddemo.py", line 24, in

asyncio.run(main())

File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/runners.py", line 43, in run

return loop.run_until_complete(main)

File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 589, in run_until_complete

return future.result()

asyncio.exceptions.CancelledError

下面说一下代码中的逻辑,在main协程中将coro协程封装为任务real_task，然后cancel_it方法做了一个取消任务的逻辑some_task.cancel()。并打印一句话。然后通过await去运行real_task方法，执行代码之后看到上面的结果。出现了asyncio.exceptions.CancelledError错误，同时看到coro只打印了一个开始休眠，后面的结束休眠没有打印。也就是说我们在取消一个任务的时候，里面对于的协程也被取消了。如果我们想在取消任务之后协程还能顺利执行完，就需要用到另外一个函数shield.

asyncio.shield

该方法的作用是，在执行cancel取消一个task之后，task里面的协程仍然可以执行结束，不会像上面的coro那样出现错误。

asyncio.shield(aw, *, loop=None)

aw表示需要传入一个 Task。

接下来我们就使用这个方法对上面的例子做一个修改。

从代码中体会它的作用

import asyncio

async def coro():

print('开始休眠')

await asyncio.sleep(2)

print('结束休眠')

async def cancel_it(some_task):

await asyncio.sleep(0.5)

some_task.cancel()

print('取消some_task任务')

async def main():

real_task = asyncio.create_task(coro())

shield = asyncio.shield(real_task)

await cancel_it(shield)

await real_task

if __name__ == '__main__':

asyncio.run(main())

运行之后的结果

开始休眠

取消some_task任务

结束休眠

可以看到尽管some_task任务被取消,但是coro仍然成功的打印了最好的“结束休眠”。通过上面的例子我想大家应该知道shield的作用了。

超时等待

有时候需要等待一个任务完成之后再进行下一个,但是有的时候并不需要运行完就返回。

这个时候可以使用wait_for

asyncio.wait_for

该方法的语法如下:

asyncio.wait_for(aw, timeout, *, loop=None)

aw是一个任务，timeout可以指定超时时间。如果发生超时，它将取消该任务并引发asyncio.TimeoutError,此时为了保证任务中协程完成可以使用上面说的 shield。

import asyncio

async def foo():

await asyncio.sleep(1)

print("in foo")

async def eternity():

# Sleep for one hour

await foo()

await asyncio.sleep(3600)

print('yay!')

async def main():

# Wait for at most 1 second

try:

await asyncio.wait_for(asyncio.shield(eternity()), timeout=1.0)

except asyncio.TimeoutError:

print('timeout!')

if __name__ == '__main__':

asyncio.run(main())

输出

in foo

timeout!

按照上面的经验可知道，如果我们把asyncio.shield去掉之后，“in foo”就无法输出了。

协程配合线程

asyncio.run_coroutine_threadsafe

该方法的语法如下:

asyncio.run_coroutine_threadsafe(coro, loop)

其实在协程中也可以使用多线程,有时候我们需要在主线程中启动一个子线程去做别的任务，这个时候我们就要用到下面的方法了，先上一个流畅的Python中的代码。

import time

import asyncio

from threading import Thread

now = lambda: time.time()

def start_loop(loop):

asyncio.set_event_loop(loop)

loop.run_forever()

async def do_some_work(x):

print(f'Waiting {x}')

await asyncio.sleep(x)

print(f'Done after {x}s')

def more_work(x):

print(f'More work {x}')

time.sleep(x)

print('Finished more work {x}')

start = now()

# 主线程中创建一个 new_loop

new_loop = asyncio.get_event_loop()

# 创建子线程 在其中开启无限事件循环

t = Thread(target=start_loop, args=(new_loop,))

t.start()

print(f'TIME: {time.time() - start}')

# 在主线程中新注册协程对象

# 这样即可在子线程中进行事件循环的并发操作 同时主线程又不会被 block

# 一共执行的时间大概在 6 s 左右

asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)

asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)

上述的例子，主线程中创建一个new_loop，然后在另外的子线程中开启一个无限事件循环。主线程通过run_coroutine_threadsafe新注册协程对象。这样就能在子线程中进行事件循环的并发操作，同时主线程又不会被block。一共执行的时间大概在6s左右。

同步原语

尽管asyncio应用通常作为单线程运行，不过仍被构建为并发应用。由于I/O以及其他外部事件的延迟和中断，每个协程或任务可能按一种不可预知的顺序执行。为了支持安全的并发执行，asyncio包含了threading和multiprocessing模块中的一些底层原语的实现。

这里介绍两个经常用到的例子

队列(Queue)

asyncio.Queue为协程提供了一个先进先出的数据结构,这与线程的queue.Queue或进程的multiprocessing.Queue很类似，下面先看一个简单的例子，它是一个非阻塞的队列。

import asyncio

from asyncio import Queue

queue=Queue()

async def start():

[queue.put_nowait(i) for i in range(1, 10)]

await asyncio.create_task(work()) #put_nowait表示放入元素

async def work():

try:

while not queue.empty():#判断队列的元素是否为空

num = queue.get_nowait()#获取元素

print(f"获取数字:{num}")

queue.task_done()#告诉队列该任务处理完。

except asyncio.CancelledError:

pass

if __name__ == '__main__':

asyncio.run(start())

输出结果:

获取数字:1

获取数字:2

获取数字:3

获取数字:4

获取数字:5

获取数字:6

获取数字:7

获取数字:8

获取数字:9

在做爬虫的时候对于url的处理，经常会用到队列的操作。另外一个要说的同步原语就是信号量。

信号量(Semaphore)

简单说下什么是信号量，我们用停车场和车进行比喻。一个停车场一共就5个车位，所以我们知道可以同时容纳最多5辆车,这五个车位就是信号量。

然后说信号量的行为，当有车离开停车场的时候外面的车就会进来补，比如有2辆车离开，那么就可以再进来2辆车，依次类推，上面这个过程就是描述了信号量这个东西。下面我们看如何在程序中使用。

asyncio.Semaphore模块就是一个维持并发量的模块，我们用它起到一个限流的效果。首先来一段代码。

import asyncio

sem=asyncio.Semaphore(3) #信号量指定为3

async def branch(num):

async with sem: #通过异步上下文关键子控制并发量

print(f"获取当前数字:{num}")

await asyncio.sleep(0.5)

async def main():

tasks=[asyncio.create_task(branch(i)) for i in range(10)] #将协程封装成任务共10个

await asyncio.wait(tasks) #执行这些任务

if __name__ == '__main__':

asyncio.run(main())

执行之后你会发现

获取当前数字:0

获取当前数字:1

获取当前数字:2

Task exception was never retrieved

future: exception=RuntimeError("Task cb=[_wait.._on_completion() at /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/tasks.py:478]> got Future attached to a different loop")>

关键就是 attached to a different loop,这个地方说是当前的事件循环发生了改变,这个问题在Python3.6的时候是不会出现的。

为什么3.8出错了,这是因为

我的信号量没有在循环内创建。也就是在asyncio.run()创建的循环之外创建了它们，因此它们使用events.get_event_loop()这就导致了新的事件循环产生。 asyncio.run()创建一个新循环，然后在一个循环中创建的future不能在另一个循环中使用。所以问题就明确了我们需要在循环之内创建。也就是我们需要定义一个全局变量，然后在主循环内部给其赋值，看到这，可能大家想到了global，Python 3.7 增加了上下文变量 Context Variables，至于为什么不用全局变量，因为可能会被其他协程修改，不安全，在这里也可以使用。

所以我们的代码变成了下面这个样子

import asyncio

from contextvars import ContextVar

concurrent=ContextVar("concurrent")#定义全局上下文管理器

async def branch(num):

sem=concurrent.get()#获取上下文关键字

async with sem:

print(f"获取当前数字:{num}")

await asyncio.sleep(0.5) #为了看到明显的效果

async def main():

concurrent.set(asyncio.Semaphore(3)) #上下文管理器赋值

tasks=[asyncio.create_task(branch(i)) for i in range(10)]

await asyncio.wait(tasks)

if __name__ == '__main__':

asyncio.run(main())

然后我们再次输出

获取当前数字:0

获取当前数字:1

获取当前数字:2

获取当前数字:3

获取当前数字:4

获取当前数字:5

获取当前数字:6

获取当前数字:7

获取当前数字:8

获取当前数字:9

可以看到程序每隔3组输出一次，这就达到了我们想要的效果了。

后记

到目前为止，asyncio 常用的操作就是上面这些了，关于更多的asyncio用法大家可以参考官网的api文档学习，同时学习这些内容，也是为了我的后续文章aiohttp异步爬虫模块打下基础，希望大家能够多动手实践，加深对代码的影响。后续我将和大家一起走入异步爬虫aiohttp的实践部分，敬请期待！