对python asyncio的一些理解和简单使用

2020-12-19 04:06:34 编辑 0 阅读量:616

协程 (coroutine) 几乎是 Python 里最为复杂的特性之一了，这篇文章我们来说一说 asyncio 的内部实现机制，借此来理解一门语言要支持协程需要做的工作。

本文需要提前了解 Python 的 yeild from 语法，不了解的话，可以看看 之前关于 Generator 的文章 ；另外，最好对 future/promise 的概念有一定了解。文中前半部分不会介绍如何使用 asyncio 及协程，并且文中第一部分给出的代码不一定能实际运行（不然代码量太大）。

# 一、协程的理解
## 1.1 多线程与协程
CPU 的执行是顺序的，线程是操作系统提供的一种机制，允许我们在操作系统的层面上实现“并行”。而协程则可以认为是应用程序提供的一种机制（用户或库来完成），允许我们在应用程序的层面上实现“并行”。

由于本质上程序是顺序执行的，要实现这种“并行”的假像，我们需要一种机制，来“暂停”当前的执行流，并在之后“恢复”之前的执行流。这在操作系统及多线程/多进程中称为“上下文切换” (context switch)。其中“上下文”记录了某个线程执行的状态，包括线程里用到的各个变量，线程的调用栈等。而“切换”指的就是保存某个线程当前的运行状态，之后再从之前的状态中恢复。只不过线程相关的工作是由操作系统完成，而协程则是由应用程序自己来完成。

与线程不同的时，协程完成的功能通常较小，所以会有需求将不同的协程串起来，我们暂时称它为协程链 (coroutine chain)。

那么，与线程类似，要实现一个协程的库，我们需要这几样东西：

事件循环 (event loop)。一方面，它类似于 CPU ，顺序执行协程的代码；另一方面，它相当于操作系统，完成协程的调度，即一个协程“暂停”时，决定接下来执行哪个协程。
上下文的表示。在 Python 中，我们使用 Python 本身支持的生成器 Generator 来代表基本的上下文，但协程链是如何工作的呢？
上下文的切换。最基础的切换也是通过 Python 生成器的 yeild 加强版语法来完成的，但我们还要考虑协程链的情况。

## 1.2 Event Loop
首先，因为协程是一种能暂停的函数，那么它暂停是为了什么？一般是等待某个事件，比如说某个连接建立了；某个 socket 接收到数据了；某个计时器归零了等。而这些事件应用程序只能通过轮询的方式得知是否完成，但是操作系统（所有现代的操作系统）可以提供一些中断的方式通知应用程序，如 select, epoll, kqueue 等等。

那么有了操作系统的支持，我们就可以手写这样的循环（伪代码）：
```
while True
    happend = poll_events(events_to_listen, timeout)
    process_events(happend)
```
第一个问题是：如何注册我们想监听的事件？很简单，把事件加到 events_to_listen 里就可以了。第二个问题，可以监听什么事件？由于 process_events 需要操作系统的支持，那么我们想监听的事件是需要操作系统支持才行的，一般操作系统支持网络 I/O 的文件描述符 (file descriptor)。

接下来，当事件发生时，我们要指定做一些事，一般称为回调 (callback)。也就是说我们需要告诉 event loop 一个 事件:回调 的对应关系。现在我们把 event loop 用类表示：
```
class EventLoop:
    def __init__(self):
        self.events_to_listen = []
        self.callbacks = {}
        self.timeout = None

def register_event(self, event, callback):
        self.events_to_listen.append(event)
        self.callbacks[event] = callback

def unregister_event(self, event):
        self.events_to_listen.remove(evenrt)
        del self.callbacks[event]

def _process_events(self, events):
        for event in events:
            self.callbacks[event](event)

def start_loop(self):
        while True:
            events_happend = poll_events(self.events_to_listen, timeout)
            self._process_events(events_happend)

loop = EventLoop()
loop.register_event(fd, callback)
loop.start_loop()
```
register_event 用到注册 事件: 回调 的关系，start_loop 用于开启事件循环。

现在，你不是想说，之前提到过事件也包括“某个计时器归零了”，但 poll_events 只支持网络 I/O 的文件描述符，计时器又要如何实现呢？一般 poll_events 函数是支持 timeout 参数表示等待的时间。因此，可以修改 start_loop:
```
def call_later(self, delay, callback):
    self.call_at(now() + delay, callback)

def call_at(self, when, callback):
    self.timeout_callbacks[when] = callback

def start_loop(self):
    while True:
        timeout = min(self.timeout_callbacks.keys()) - now()
        events_happend = poll_events(self.events_to_listen, timeout)
        if not empty(events_happend):
            self._process_events(events_happend)
        self._process_timeout_events()

def _process_timeout_events(self):
    time_now = now()
    for time, callback in self.timeout_callbacks.iteritems():
        if time < time_now:
            callback()
            del self.timeout_callbacks[time]
```
这里 poll_events 之前，会去计算所有计时器事件最少需要等待的时间，这个时间内即使没有事件发生，poll_events 也会退出，以便触发计时器事件。 _process_timeout_events 函数的作用是对比当前时间与计时器的目标执行时间，如果目标执行时间已经到达，则执行相应的回调函数。

于是一个简单的 event loop 就完成了。可以看到，它是异步操作的基础：允许等待某个事件的发生并执行相应的操作。同时，它还是个简单的调度器，能顺序地执行发生事件的回调函数。

## 1.3 Callback vs Promise vs await
好了，现在我们有了 event loop ，它允许我们为事件注册回调函数。现在假设我们要顺序调用几个 API， 用阻塞式编程如下：
```
result1 = api1()
result2 = api2(result1)
result3 = api3(result2)
...
```
如果这几个 API 都是异步的，用 event loop + callback 怎么实现？
```
# Implementation for api
def api1(callback):
    def callback_for_api1():
        result1 = some_calculation_1()
        event_loop.unregister_event(event1)
        return callback(result1)
    event_loop.register_event(event1, callback_for_api1)

def api2(result, callback):
    def callback_for_api2():
        result2 = some_calculation_2(result)
        event_loop.unregister_event(event2)
        return callback(result2)
    event_loop.register_event(event2, callback_for_api2)
...

# Our code
global result
def api1_callback(result1):
    def api2_callback(result2):
        def api3_callback(result3):
            global result
            result = some_calculation(result3)
        return api3(result2, api3_callback)
    return api2(result1, api2_callback)
api1(api1_callback)
```
这里 api1 api2 的实现由于需要用 event loop 来注册注销某些事件，所以显得特别复杂，这里我们可以先忽略它们的实现，但是看最后一段“用户代码”是不是极其复杂？随着操作的复杂性增加，回调函数的嵌套会越变越深。如果你熟悉Javascript，你应该听过“callback hell”的大名。回调函数的方式为什么不好？最重要的就是它违反了我们写代码的直觉，我们都习惯顺序执行的代码。

例如上例中，我们期待的是 api1 先执行，我们再用它的结果做点什么，但采用回调的方式，我们就需要在写 api1 的回调时，就去思考我们想用它的结果做些什么操作。在这个例子里，我们需要调用 api2 及 api3，这些嵌套的思考又得一遍遍重复下去。最终代码非常难以理解。

因此 Javascript 提出了 Promise ，所谓的 promise 像是一个占位符，它表示一个运算现在还未完成，但我保证它会做完的；你可以指定它完成的时候做些其它的事。下面我们尝试用这个思路去做一些改进（Python 没有原生的 promise 支持）：
```
class Promise():
    def __init__(self):
        pass
    def then(self, callback_that_return_promise):
        self._then = callback_that_return_promise
    def set_result(self, result):
        return self._then(result)

# Implementation for api
def api1():
    promise = Promise()
    def callback_for_api1():
        promise.set_result(some_calculation_1())
        event_loop.unregister_event(event1)
    event_loop.register_event(event1, callback_for_api1)
    return promise

def api2(result):
    promise = Promise()
    def callback_for_api2():
        promise.set_result((some_calculation_2(result))
        event_loop.unregister_event(event2)
        return callback(result2)
    return promise
...

# Our code
global result
promise = api1().then(lambda result1: return api2(result1))
    .then(lambda result2: return api3(result3))
    .then(lambda result3: global result; result = result3)

promise.wait_till_complete()
```
这里我们简单实现了一个我们自己的 Promise 类，当它的 set_result 方法被调用时，Promise 会去执行之前用 .then 注册的回调函数，该回调函数将执行另一些操作并返回一个新的 Promise。也因此，我们可以不断地调用 then 将不同的 Promise 组合起来。可以看到，现在我们的代码就是线性的了！

然而故事还没有结束，人们依旧不满于 Promise 的写法和用法，又提出了 async/await 的写法。在 Python 中，上面的代码用 async/await 重写如下：
```
result1 = await api1()
result2 = await api2(result1)
result3 = await api3(result2)
```
是不是简单明了？它的效果和我们前几个例子是等价的，但它的写法与我们初开始的阻塞版本几乎一致。这样能把异步与同步的编码在结构上尽量统一起来。

这里我不禁想问，为什么大家没有一开始就想到 async/await 的方式呢？我的一个假设是 async/await 是需要语言本身的支持的，而写编译器/解释器的专家不一定有编写应用的丰富经验，是很可能从一开始就拒绝这样的修改的。因此程序员们只能自己用库的形式添加支持了。当然这纯粹是猜测，只想感叹下不同领域的隔阂。

总而言之，有了 event loop 我们就能通过回调函数来完成异步编程，但这种方式非常不友好，因此人们又提出了类似 Promise 的思想，让我们能顺序编写异步代码，最后通过语言对 async/await 的语法支持，异步与同步代码的结构就几乎达到统一。这种统一有很重要的意义，它使我们能以同步的思维去理解异步的代码而不受回调方式的代码结构的影响。

而这一切都是为了将不同的异步函数“链接”起来，只不过是 async/await 的方式最为方便。对比线程，操作系统是没有提供方式将不同的线程链接起来的，因此这种将不同的协程链接起来的工具是协程比线程好的一个方面。

## 1.4 上下文切换（恢复控制流）
前面提到过，如果某个协程在等待某些资源，我们需要暂停它的执行，在 event loop 中注册这个事件，以便当事件发生的时候，能再次唤醒该协程的执行。

这里举一个 Python 官方文档 的例子：
```
import asyncio

async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y

async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()
```
上面的代码的执行流程是：
![](http://cblog.xyz/media/2018/01/tulip_coro.png\)

这里有两个问题：
1. 谁向 event loop 注册了事件（及回调）？
2. 程序从哪里恢复执行？
程序从 print_sum 开始执行，执行到 asyncio.sleep 时需要暂停，那么肯定是在 sleep 中向 event loop 注册了计时器事件。那们问题来了，当程序恢复执行时，它应该从哪里恢复呢？

从上面的流程图中，可以看见它是从 print_sum 开始恢复，但这样的话，sleep 注册事件时就需要知道是谁（即 print_sum）调用了它，这样才能在 callback 中指定从 print_sum 开始恢复执行！

但如果不是从 print_sum 恢复执行，那么一样的，从 sleep 恢复执行后，sleep 需要知道接下来返回到什么位置（即 compute 函数中的 await 位置）， asyncio 又是如何做到这点的？

那么事实（代码实现）是怎样的呢？

当我们把一个协程用 loop.run_until_complete （或其它相似方法）执行时， event loop 会把它包裹成一个 Task 。当协程开始执行或被唤醒时，Task 的 `_step` 方法会被调用，这里 它会调用 coro.send(None) 来执行/唤醒它包裹着的协程。
```
if exc is None:
    # We use the `send` method directly, because coroutines
    # don't have `__iter__` and `__next__` methods.
    result = coro.send(None)
else:
    result = coro.throw(exc)
```
注意到这里将 coro.send 的结果赋值给了 result，那么它会返回什么呢？在我们这个例子中，协程链的最末尾是 asyncio.sleep，我们看看 它的实现：
```
@coroutine
def sleep(delay, result=None, *, loop=None):
    """Coroutine that completes after a given time (in seconds)."""
    if delay == 0:
        yield
        return result

if loop is None:
        loop = events.get_event_loop()
    future = loop.create_future()
    h = future._loop.call_later(delay,
                                futures._set_result_unless_cancelled,
                                future, result)
    try:
        return (yield from future)
    finally:
        h.cancel()
```
这里它创建了一个 future 并为它注册了事件（call_later），最终调用了 yield from future 返回。它代表什么呢？我们已经假设你明白 yield from 的使用方法，这代表 Python 会首先调用 `future.__iter__` 函数，我们来看看它长什么样：
```
def __iter__(self):
    if not self.done():
        self._asyncio_future_blocking = True
        yield self  # This tells Task to wait for completion.
    assert self.done(), "yield from wasn't used with future"
    return self.result()  # May raise too.

if compat.PY35:
    __await__ = __iter__ # make compatible with 'await' expression
```
注意这里的 yield self！也就是说 future 在第一次执行到这里时，会暂停执行并返回它自己，由于 coroutine 中使用的都是 yield from/await （它们在接收的参数上有区别，但在本文的讨论中没有区别），因此这个值会一直向上传递，到 `Task._step` 函数的 result = coro.send(None) 这里，那我们来看看 Task 对 result 做了什么，重要的是这一句：
`result.add_done_callback(self._wakeup)`
也就是说 task(print_sum) 得到了最内层暂停的 sleep 生成的 future 并为该 future 注册了一个回调，使得在 future.set_result 被调用时，`task._wakeup` 会被调用。这部分的逻辑可以看这里。

我们再回过头来看看 future.set_result 会在什么时候被调用，在 asyncio.sleep 函数里，我们为 event loop 注册了一个回调函数：
```
h = future._loop.call_later(delay,
                            futures._set_result_unless_cancelled,
                            future, result)
```
那么这个 _set_result_unless_cancelled 是这样的：
```
def _set_result_unless_cancelled(fut, result):
    """Helper setting the result only if the future was not cancelled."""
    if fut.cancelled():
        return
    fut.set_result(result)
```
因此，所有的流程应该是这样的：
![](http://cblog.xyz/media/2018/01/asyncio-flow.png\)

小结
那么 asyncio 做为一个库，做了什么，没做什么？
- 控制流的暂停与恢复，这是通过 Python 内部的 Generator（生成器）相关的功能实现的。
- 协程链，即把不同协程链链接在一起的机制。依旧是通过 Python 的内置支持，即 async/await，或者说是生成器的 yield from。
- Event Loop，这个是 asyncio 实现的。它决定了我们能对什么事件进行异步操作，目前只支持定时器与网络 IO 的异步。
- 协程链的控制流恢复，即内部的协程暂停了，恢复时却需要从最外层的协程开始恢复。这是 asyncio 实现的内容。
- 其它的库支持，这里指的是像 asyncio.sleep() 这种协程链的最内层的协程，因此我们一般不希望自己去调用 event loop 注册/注销事件。

因此，如果没有 asyncio，我们要实现相应的功能，主要的内容就是 Event Loop 及控制流的恢复，最后再加上一些好用的协程函数。

# 二、asyncio的简单使用
这里主要有一下几个概念
- event_loop 事件循环：程序开启一个无限的循环，程序员会把一些函数注册到事件循环上。当满足事件发生的时候，调用相应的协程函数。
- coroutine 协程：协程对象，指一个使用async关键字定义的函数，它的调用不会立即执行函数，而是会返回一个协程对象。协程对象需要注册到事件循环，由事件循环调用。
- task 任务：一个协程对象就是一个原生可以挂起的函数，任务则是对协程进一步封装，其中包含任务的各种状态。
- future： 代表将来执行或没有执行的任务的结果。它和task上没有本质的区别
- async/await 关键字：python3.5 用于定义协程的关键字，async定义一个协程，await用于挂起阻塞的异步调用接口。

## 2.1 定义一个协程

定义一个协程很简单，使用async关键字，就像定义普通函数一样：

```
import time
import asyncio

now = lambda : time.time()

async def do_some_work(x):
    print('Waiting: ', x)

start = now()

coroutine = do_some_work(2)

loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)

print('TIME: ', now() - start)
```

通过async关键字定义一个协程（coroutine），协程也是一种对象。协程不能直接运行，需要把协程加入到事件循环（loop），由后者在适当的时候调用协程。asyncio.get_event_loop方法可以创建一个事件循环，然后使用run_until_complete将协程注册到事件循环，并启动事件循环。因为本例只有一个协程，于是可以看见如下输出：
```
Waiting:  2
TIME:  0.0004658699035644531
```

## 2.2 创建一个task

协程对象不能直接运行，在注册事件循环的时候，其实是run_until_complete方法将协程包装成为了一个任务（task）对象。所谓task对象是Future类的子类。保存了协程运行后的状态，用于未来获取协程的结果。

```
import asyncio
import time

now = lambda : time.time()

async def do_some_work(x):
    print('Waiting: ', x)

start = now()

coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
# task = asyncio.ensure_future(coroutine)
task = loop.create_task(coroutine)
print(task)
loop.run_until_complete(task)
print(task)
print('TIME: ', now() - start)
```
可以看到输出结果为：

```
<Task pending coro=<do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:17>>
Waiting:  2
<Task finished coro=<do_some_work() done, defined at /Users/ghost/Rsj217/python3.6/async/async-main.py:17> result=None>
TIME:  0.0003490447998046875
```
创建task后，task在加入事件循环之前是pending状态，因为do_some_work中没有耗时的阻塞操作，task很快就执行完毕了。后面打印的finished状态。

asyncio.ensure_future(coroutine) 和 loop.create_task(coroutine)都可以创建一个task，run_until_complete的参数是一个futrue对象。当传入一个协程，其内部会**自动封装成task**，task是Future的子类。isinstance(task, asyncio.Future)将会输出True。

## 2.3 绑定回调

绑定回调，在task执行完毕的时候可以获取执行的结果，回调的最后一个参数是future对象，通过该对象可以获取协程返回值。如果回调需要多个参数，可以通过偏函数导入。

```
import time
import asyncio

now = lambda : time.time()

async def do_some_work(x):
    print('Waiting: ', x)
    return 'Done after {}s'.format(x)

def callback(future):
    print('Callback: ', future.result())

start = now()

coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)
loop.run_until_complete(task)

print('TIME: ', now() - start)
```

```
def callback(t, future):
    print('Callback:', t, future.result())
 
task.add_done_callback(functools.partial(callback, 2))
```
可以看到，coroutine执行结束时候会调用回调函数。并通过参数future获取协程执行的结果。我们创建的task和回调里的future对象，实际上是同一个对象。

## 2.4 future 与 result

回调一直是很多异步编程的恶梦，程序员更喜欢使用同步的编写方式写异步代码，以避免回调的恶梦。回调中我们使用了future对象的result方法。前面不绑定回调的例子中，我们可以看到task有fiinished状态。在那个时候，可以直接读取task的result方法。
```
async def do_some_work(x):
    print('Waiting {}'.format(x))
    return 'Done after {}s'.format(x)

start = now()

coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)
loop.run_until_complete(task)

print('Task ret: {}'.format(task.result()))
print('TIME: {}'.format(now() - start))
```
可以看到输出的结果：
```
Waiting:  2
Task ret:  Done after 2s
TIME:  0.0003650188446044922
```

## 2.5 阻塞和await

使用async可以定义协程对象，使用await可以针对耗时的操作进行挂起，就像生成器里的yield一样，函数让出控制权。协程遇到await，事件循环将会挂起该协程，执行别的协程，直到其他的协程也挂起或者执行完毕，再进行下一个协程的执行。

耗时的操作一般是一些IO操作，例如网络请求，文件读取等。我们使用asyncio.sleep函数来模拟IO操作。协程的目的也是让这些IO操作异步化。
```
import asyncio
import time

now = lambda: time.time()

async def do_some_work(x):
    print('Waiting: ', x)
    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)

start = now()

coroutine = do_some_work(2)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(coroutine)
loop.run_until_complete(task)

print('Task ret: ', task.result())
print('TIME: ', now() - start)
```
在 sleep的时候，使用await让出控制权。即当遇到阻塞调用的函数的时候，使用await方法将协程的控制权让出，以便loop调用其他的协程。现在我们的例子就用耗时的阻塞操作了。

## 2.6 并发和并行

并发和并行一直是容易混淆的概念。并发通常指有多个任务需要同时进行，并行则是同一时刻有多个任务执行。用上课来举例就是，并发情况下是一个老师在同一时间段辅助不同的人功课。并行则是好几个老师分别同时辅助多个学生功课。简而言之就是一个人同时吃三个馒头还是三个人同时分别吃一个的情况，吃一个馒头算一个任务。

asyncio实现并发，就需要多个协程来完成任务，每当有任务阻塞的时候就await，然后其他协程继续工作。创建多个协程的列表，然后将这些协程注册到事件循环中。

```
import asyncio

import time

now = lambda: time.time()

async def do_some_work(x):
    print('Waiting: ', x)

await asyncio.sleep(x)
    return 'Done after {}s'.format(x)

start = now()

coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)

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

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

for task in tasks:
    print('Task ret: ', task.result())

print('TIME: ', now() - start)
```
总时间为4s左右。4s的阻塞时间，足够前面两个协程执行完毕。如果是同步顺序的任务，那么至少需要7s。此时我们使用了aysncio实现了并发。asyncio.wait(tasks) 也可以使用 asyncio.gather(*tasks) ,前者接受一个task列表，后者接收一堆task。

## 2.7 协程嵌套

使用async可以定义协程，协程用于耗时的io操作，我们也可以封装更多的io操作过程，这样就实现了嵌套的协程，即一个协程中await了另外一个协程，如此连接起来。

```
import asyncio

import time

now = lambda: time.time()

async def do_some_work(x):
    print('Waiting: ', x)

await asyncio.sleep(x)
    return 'Done after {}s'.format(x)

async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

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

dones, pendings = await asyncio.wait(tasks)

for task in dones:
        print('Task ret: ', task.result())

start = now()

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

print('TIME: ', now() - start)
```
如果使用的是 asyncio.gather创建协程对象，那么await的返回值就是协程运行的结果。
```
    results = await asyncio.gather(*tasks)

for result in results:
        print('Task ret: ', result)
```
不在main协程函数里处理结果，直接返回await的内容，那么最外层的run_until_complete将会返回main协程的结果。
```
async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(2)

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

return await asyncio.gather(*tasks)

start = now()

loop = asyncio.get_event_loop()
results = loop.run_until_complete(main())

for result in results:
    print('Task ret: ', result)
```
或者返回使用asyncio.wait方式挂起协程。
```
async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

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

return await asyncio.wait(tasks)

start = now()

loop = asyncio.get_event_loop()
done, pending = loop.run_until_complete(main())

for task in done:
    print('Task ret: ', task.result())
```
也可以使用asyncio的as_completed方法
```
async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3)
    ]
    for task in asyncio.as_completed(tasks):
        result = await task
        print('Task ret: {}'.format(result))

start = now()

loop = asyncio.get_event_loop()
done = loop.run_until_complete(main())
print('TIME: ', now() - start)
```
由此可见，协程的调用和组合十分灵活，尤其是对于结果的处理，如何返回，如何挂起，需要逐渐积累经验和前瞻的设计。

## 2.8 获取协程运行结果的一些方式
```
# coding: utf-8

import asyncio
import aiohttp

url = 'http://www.sina.com'

async def getIndex(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

loop = asyncio.get_event_loop()

#### first method to get result
# tasks = []
# for i in range(10):
    # tasks.append(getIndex(url))
# futures = asyncio.gather(*tasks)
# loop.run_until_complete(futures)
# results = futures.result()
# print results[0]

### second method to get result
# tasks = []
# for i in range(10):
    # tasks.append(asyncio.ensure_future(getIndex(url)))
# loop.run_until_complete(asyncio.wait(tasks))
# print(tasks[0].result())

### third method to get result
# tasks = []
# for i in range(10):
    # tasks.append(asyncio.ensure_future(getIndex(url)))
# results = loop.run_until_complete(asyncio.wait(tasks))
# print(results[1])

### fourth method to get result, 推荐
tasks = []
for i in range(10):
    tasks.append(getIndex(url))
results = loop.run_until_complete(asyncio.gather(*tasks))
print(results[0])

### can't do
# futures = asyncio.run_coroutine_threadsafe(asyncio.wait(tasks), loop)
# print(futures.result()[0])
```

# 三、 其他
## 3.1 coroutine 与 Future 的关系
看起来两者是一样的，因为都可以用以下的语法来异步获取结果，
```
result = await future
result = await coroutine
```
实际上，coroutine 是生成器函数，它既可以从外部接受参数，也可以产生结果。使用 coroutine 的好处是，我们可以暂停一个函数，然后稍后恢复执行。比如在涉及到网路操作的情况下，能够停下函数直到响应到来。在停下的这段时间内，我们可以切换到其他任务继续执行。
而 Future 更像是 Javascript 中的 Promise 对象。它是一个占位符，其值会在将来被计算出来。在上述的例子中，当我们在等待网络 IO 函数完成时，函数会给我们一个容器，Promise 会在完成时填充该容器。填充完毕后，我们可以用回调函数来获取实际结果。
Task 对象是 Future 的子类，它将 coroutine 和 Future 联系在一起，将 coroutine 封装成一个 Future 对象。
一般会看到两种任务启动方法，
```
tasks = asyncio.gather(
  asyncio.ensure_future(func1()),
  asyncio.ensure_future(func2())
)
loop.run_until_complete(tasks)
```
和
```
tasks = [
  asyncio.ensure_future(func1()),
  asyncio.ensure_future(func2())
  ]
loop.run_until_complete(asyncio.wait(tasks))
```
ensure_future 可以将 coroutine 封装成 Task。asyncio.gather 将一些 Future 和 coroutine 封装成一个 Future。
asyncio.wait 则本身就是 coroutine。
run_until_complete 既可以接收 Future 对象，也可以是 coroutine 对象，
```
BaseEventLoop.run_until_complete(future)
 
Run until the Future is done.
If the argument is a coroutine object, it is wrapped by ensure_future().
Return the Future's result, or raise its exception.
```

## 3.2 Task 任务的正确退出方式
在 asyncio 的任务循环中，如果使用 CTRL-C 退出的话，即使捕获了异常，Event Loop 中的任务会报错，出现如下的错误，
```
Task was destroyed but it is pending!
task: <Task pending coro=<kill_me() done, defined at test.py:5> wait_for=<Future pending cb=[Task._wakeup()]>>
```
根据官方文档，Task 对象只有在以下几种情况，会认为是退出，
`a result / exception are available, or that the future was cancelled`
Task 对象的 cancel 和其父类 Future 略有不同。当调用 Task.cancel() 后，对应 coroutine 会在事件循环的下一轮中抛出 CancelledError 异常。使用 Future.cancelled() 并不能立即返回 True（用来表示任务结束），只有在上述异常被处理任务结束后才算是 cancelled。
故结束任务可以用
```
for task in asyncio.Task.all_tasks():
  task.cancel()
```
这种方法将所有任务找出并 cancel。
但 CTRL-C 也会将事件循环停止，所以有必要重启事件循环，
```
try:
  loop.run_until_complete(tasks)
except KeyboardInterrupt as e:
  for task in asyncio.Task.all_tasks():
    task.cancel()
  loop.run_forever() # restart loop
finally:
  loop.close()
```
在每个 Task 中捕获异常是必要的，如果不确定，可以使用
`asyncio.gather(..., return_exceptions=True)`
将异常转换为正常的结果返回。

总之，协程是一个比较复杂的功能，要想更好的利用，还需要进一步的熟悉

---
参考：
- [1][理解 Python asyncio](http://lotabout.me/2017/understand-python-asyncio/)
- [2][Python黑魔法 --- 异步IO（ asyncio） 协程](http://python.jobbole.com/87310/)
- [3][18.5.3. Tasks and coroutines](https://docs.python.org/3/library/asyncio-task.html)
- [4][Python协程：从yield/send到async/await](http://python.jobbole.com/86069/)
- [5][asyncio 的 coroutine对象 与 Future对象使用指南](http://www.jb51.net/article/92397.htm)

对python asyncio的一些理解和简单使用

附件: