python併發2：使用asyncio處理併發

asyncio

asyncio 是Python3.4 以後引入的標準庫的，這個包使用事件循環驅動的協程實現併發。
asyncio 包在引入標準庫以前代號 「Tulip」（鬱金香），因此在網上搜索資料時，會常常看到這種花的名字。

什麼是事件循環?

wiki 上說：事件循環是」一種等待程序分配事件或者消息的編程架構「。基本上來講事件循環就是：」當A發生時，執行B"。或者用最簡單的例子來解釋這一律念就是每一個瀏覽器中都存在的JavaScript事件循環。當你點擊了某個東西（「當A發生時」），這一點擊動做會發送給JavaScript的事件循環，並檢查是否存在註冊過的onclick 回調來處理這一點擊（執行B)。只要有註冊過的回調函數就會伴隨點擊動做的細節信息被執行。事件循環被認爲是一種虛幻是由於它不停的手機事件並經過循環來發如何應對這些事件。

對 Python 來講，用來提供事件循環的 asyncio 被加入標準庫中。asyncio 重點解決網絡服務中的問題，事件循環在這裏未來自套接字（socket）的 I/O 已經準備好讀和/或寫做爲「當A發生時」（經過selectors模塊）。除了 GUI 和 I/O，事件循環也常常用於在別的線程或子進程中執行代碼，並將事件循環做爲調節機制（例如，合做式多任務）。若是你剛好理解 Python 的 GIL，事件循環對於須要釋放 GIL 的地方頗有用。

線程與協程

咱們先看兩斷代碼，分別用 threading 模塊和asyncio 包實現的一段代碼。

# sinner_thread.py

import threading
import itertools
import time
import sys


class Signal: # 這個類定義一個可變對象，用於從外部控制線程
    go = True


def spin(msg, signal):  # 這個函數會在單獨的線程中運行，signal 參數是前邊定義的Signal類的實例
    write, flush = sys.stdout.write, sys.stdout.flush
    for char in itertools.cycle('|/-\\'):  # itertools.cycle 函數從指定的序列中反覆不斷地生成元素
        status = char + ' ' + msg
        write(status)
        flush()
        write('\x08' * len(status))  # 使用退格符把光標移回行首
        time.sleep(.1)  # 每 0.1 秒刷新一次
        if not signal.go:  # 若是 go屬性不是 True，退出循環
            break

    write(' ' * len(status) + '\x08' * len(status))  # 使用空格清除狀態消息，把光標移回開頭


def slow_function():  # 模擬耗時操做
    # 僞裝等待I/O一段時間
    time.sleep(3)  # 調用sleep 會阻塞主線程，這麼作事爲了釋放GIL，建立從屬線程
    return 42


def supervisor():  # 這個函數設置從屬線程，顯示線程對象，運行耗時計算，最後殺死進程
    signal = Signal()
    spinner = threading.Thread(target=spin,
                               args=('thinking!', signal))
    print('spinner object:', spinner)  # 顯示線程對象 輸出 spinner object: <Thread(Thread-1, initial)>
    spinner.start()  # 啓動從屬進程
    result = slow_function()  # 運行slow_function 行數，阻塞主線程。同時叢書線程以動畫形式旋轉指針
    signal.go = False
    spinner.join()  # 等待spinner 線程結束
    return result

def main():
    result = supervisor()  
    print('Answer', result)


if __name__ == '__main__':
    main()

執行一下，結果大體是這個樣子：

這是一個動圖，「thinking" 前的 線是會動的（爲了錄屏，我把sleep 的時間調大了）

python 並無提供終止線程的API，因此若想關閉線程，必須給線程發送消息。這裏咱們使用signal.go 屬性：在主線程中把它設置爲False後，spinner 線程會接收到，而後退出

如今咱們再看下使用 asyncio 包的版本：

# spinner_asyncio.py
# 經過協程以動畫的形式顯示文本式旋轉指針

import asyncio
import itertools
import sys


@asyncio.coroutine # 打算交給asyncio 處理的協程要使用 @asyncio.coroutine 裝飾
def spin(msg):
    write, flush = sys.stdout.write, sys.stdout.flush
    for char in itertools.cycle('|/-\\'):  # itertools.cycle 函數從指定的序列中反覆不斷地生成元素
        status = char + ' ' + msg
        write(status)
        flush()
        write('\x08' * len(status))  # 使用退格符把光標移回行首
        try:
            yield from asyncio.sleep(0.1)  # 使用 yield from asyncio.sleep(0.1) 代替 time.sleep(.1), 這樣的休眠不會阻塞事件循環
        except asyncio.CancelledError:  # 若是 spin 函數甦醒後拋出 asyncio.CancelledError 異常，其緣由是發出了取消請求
            break

    write(' ' * len(status) + '\x08' * len(status))  # 使用空格清除狀態消息，把光標移回開頭


@asyncio.coroutine
def slow_function():  # 5 如今此函數是協程，使用休眠僞裝進行I/O 操做時，使用 yield from 繼續執行事件循環
    # 僞裝等待I/O一段時間
    yield from asyncio.sleep(3)  # 此表達式把控制權交給主循環，在休眠結束後回覆這個協程
    return 42


@asyncio.coroutine
def supervisor():  #這個函數也是協程，所以可使用 yield from 驅動 slow_function
    spinner = asyncio.async(spin('thinking!'))  # asyncio.async() 函數排定協程的運行時間，使用一個 Task 對象包裝spin 協程，並當即返回
    print('spinner object:', spinner)  # Task 對象，輸出相似 spinner object: <Task pending coro=<spin() running at spinner_asyncio.py:6>>
    # 驅動slow_function() 函數，結束後，獲取返回值。同事事件循環繼續運行，
    # 由於slow_function 函數最後使用yield from asyncio.sleep(3) 表達式把控制權交給主循環
    result = yield from slow_function()
    # Task 對象能夠取消；取消後會在協程當前暫停的yield處拋出 asyncio.CancelledError 異常
    # 協程能夠捕獲這個異常，也能夠延遲取消，甚至拒絕取消
    spinner.cancel()

    return result

def main():
    loop = asyncio.get_event_loop()  # 獲取事件循環引用
    # 驅動supervisor 協程，讓它運行完畢；這個協程的返回值是此次調用的返回值
    result = loop.run_until_complete(supervisor())
    loop.close()
    print('Answer', result)


if __name__ == '__main__':
    main()

除非想阻塞主線程，從而凍結事件循環或整個應用，不然不要再 asyncio 協程中使用 time.sleep().
若是協程須要在一段時間內什麼都不作，應該使用 yield from asyncio.sleep(DELAY)

使用 @asyncio.coroutine 裝飾器不是強制要求，但建議這麼作由於這樣能在代碼中突顯協程，若是還沒從中產出值，協程就把垃圾回收了（意味着操做未完成，可能有缺陷），能夠發出警告。這個裝飾器不會預激協程。

這兩段代碼的執行結果基本相同，如今咱們看一下兩段代碼的核心代碼 supervisor 主要區別：

• asyncio.Task 對象差很少與 threading.Thread 對象等效（Task 對象像是實現寫做時多任務的庫中的綠色線程

• Task 對象用於驅動協程，Thread 對象用於調用可調用的對象

• Task 對象不禁本身動手實例化，而是經過把協程傳給 asyncio.async(...) 函數或 loop.create_task(...) 方法獲取

• 獲取的Task 對象已經排定了運行時間；Thread 實例必須調用start方法，明確告知它運行

• 在線程版supervisor函數中，slow_function 是普通的函數，由線程直接調用，而異步版的slow_function 函數是協程，由yield from 驅動。

• 沒有API能從外部終止線程，由於線程隨時可能被中斷。而若是想終止任務，可使用Task.cancel() 實例方法，在協程內部拋出CancelledError 異常。協程能夠在暫停的yield 處捕獲這個異常，處理終止請求

• supervisor 協程必須在main 函數中由loop.run_until_complete 方法執行。

協程和線程相比關鍵的一個優勢是，
線程必須記住保留鎖，去保護程序中的重要部分，防止多步操做再執行的過程當中中斷，防止山水處於於曉狀態
協程默認會作好保護，咱們必須顯式產出（使用yield 或 yield from 交出控制權）才能讓程序的餘下部分運行。

asyncio.Future：故意不阻塞

asynci.Future 類與 concurrent.futures.Future 類的接口基本一致，不過實現方式不一樣，不可互換。

上一篇[python併發 1：使用 futures 處理併發]()咱們介紹過 concurrent.futures.Future 的 future，在 concurrent.futures.Future 中，future只是調度執行某物的結果。在 asyncio 包中，BaseEventLoop.create_task(...) 方法接收一個協程，排定它的運行時間，而後返回一個asyncio.Task 實例（也是asyncio.Future 類的實例，由於 Task 是 Future 的子類，用於包裝協程。（在 concurrent.futures.Future 中，相似的操做是Executor.submit(...)）。

與concurrent.futures.Future 相似，asyncio.Future 類也提供了

• .done() 返回布爾值，表示Future 是否已經執行

• .add_done_callback() 這個方法只有一個參數，類型是可調用對象，Future運行結束後會回調這個對象。

• .result() 這個方法沒有參數，所以不能指定超時時間。 若是調用 .result() 方法時期尚未運行完畢，會拋出 asyncio.InvalidStateError 異常。

對應的 concurrent.futures.Future 類中的 Future 運行結束後調用result(), 會返回可調用對象的結果或者拋出執行可調用對象時拋出的異常，若是是 Future 沒有運行結束時調用 f.result()方法，這時會阻塞調用方所在的線程，直到有結果返回。此時result 方法還能夠接收 timeout 參數，若是在指定的時間內 Future 沒有運行完畢，會拋出 TimeoutError 異常。

咱們使用asyncio.Future 時， 一般使用yield from，從中獲取結果，而不是使用 result()方法 yield from 表達式在暫停的協程中生成返回值，回覆執行過程。

asyncio.Future 類的目的是與 yield from 一塊兒使用，因此一般不須要使用如下方法：

• 不需調用 my_future.add_down_callback(...), 由於能夠直接把想在 future 運行結束後的操做放在協程中 yield from my_future 表達式的後邊。（由於協程能夠暫停和恢復函數）

• 無需調用 my_future.result(), 由於 yield from 產生的結果就是（result = yield from my_future)

在 asyncio 包中，可使用yield from 從asyncio.Future 對象中產出結果。這也就意味着咱們能夠這麼寫：

res = yield from foo()  # foo 能夠是協程函數，也能夠是返回 Future 或 task 實例的普通函數

asyncio.async(...)* 函數

asyncio.async(coro_or_future, *, loop=None)

這個函數統一了協程和Future: 第一個參數能夠是兩者中的任意一個。若是是Future 或者 Task 對象，就直接返回，若是是協程，那麼async 函數會自動調用 loop.create_task(...) 方法建立 Task 對象。 loop 參數是可選的，用於傳入事件循環; 若是沒有傳入，那麼async函數會經過調用asyncio.get_event_loop() 函數獲取循環對象。

BaseEventLoop.create_task(coro)

這個方法排定協程的執行時間，返回一個 asyncio.Task 對象。若是在自定義的BaseEventLoop 子類上調用，返回的對象多是外部庫中與Task類兼容的某個類的實例。

BaseEventLoop.create_task() 方法只在Python3.4.2 及以上版本可用。 Python3.3 只能使用 asyncio.async(...)函數。

若是想在Python控制檯或者小型測試腳本中實驗future和協程，可使用下面的片斷：

import asyncio

def run_sync(coro_or_future):
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(coro_or_future)

a = run_sync(some_coroutine())

使用asyncio 和 aiohttp 包下載

如今，咱們瞭解了asyncio 的基礎知識，是時候使用asyncio 來重寫咱們 上一篇 [python併發 1：使用 futures 處理併發]() 下載國旗的腳本了。

先看一下代碼：

import asyncio

import aiohttp  # 須要pip install aiohttp

from flags import save_flag, show, main, BASE_URL


@asyncio.coroutine  # 咱們知道，協程應該使用 asyncio.coroutine 裝飾
def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
     # 阻塞的操做經過協程實現，客戶代碼經過yield from 把指責委託給協程，以便異步操做
    resp = yield from aiohttp.request('GET', url) 
    # 讀取也是異步操做
    image = yield from resp.read()
    return image


@asyncio.coroutine
def download_one(cc):  # 這個函數也必須是協程，由於用到了yield from
    image = yield from get_flag(cc) 
    show(cc)
    save_flag(image, cc.lower() + '.gif')
    return cc


def download_many(cc_list):
    loop = asyncio.get_event_loop()  # 獲取事件序號底層實現的引用
    to_do = [download_one(cc) for cc in sorted(cc_list)] # 調用download_one 獲取各個國旗，構建一個生成器對象列表
    # 雖然函數名稱是wait 但它不是阻塞型函數，wait 是一個協程，等傳給他的全部協程運行完畢後結束
    wait_coro = asyncio.wait(to_do)
    res, _ = loop.run_until_complete(wait_coro) # 執行事件循環，知道wait_coro 運行結束；事件循環運行的過程當中，這個腳本會在這裏阻塞。
    loop.close() # 關閉事件循環
    return len(res)

if __name__ == '__main__':
    main(download_many)

這段代碼的運行簡述以下：

1. 在download_many 函數獲取一個事件循環，處理調用download_one 函數生成的幾個協程對象

2. asyncio 事件循環一次激活各個協程

3. 客戶代碼中的協程（get_flag）使用 yield from 把指責委託給庫裏的協程（aiohttp.request)時，控制權交還給事件循環，執行以前排定的協程

4. 事件循環經過基於回調的底層API，在阻塞的操做執行完畢後得到通知。

5. 得到通知後，主循環把結果發給暫停的協程

6. 協程向前執行到下一個yield from 表達式，例如 get_flag 函數的yield from resp.read()。事件循環再次獲得控制權，重複第4~6步，直到循環終止。

download_many 函數中，咱們使用了 asyncio.wait(...) 函數，這個函數是一個協程，協程的參數是一個由future或者協程構成的可迭代對象；wait 會分別把各個協程包裝進一個Task對象。最終的結果是，wait 處理的全部對象都經過某種方式變成Future 類的實例。

wait 是協程函數，所以，返回的是一個協程或者生成器對象；waite_coro 變量中存儲的就是這種對象

loop.run_until_complete 方法的參數是一個future 或協程。若是是協程，run_until_complete 方法與 wait 函數同樣，把協程包裝進一個Task 對象中。這裏 run_until_complete 方法把 wait_coro 包裝進一個Task 對象中，由yield from 驅動。wait_coro 運行結束後返回兩個參數，第一個參數是結束的future 第二個參數是未結束的future。

<section class="caption">wait</section>有兩個命名參數，timeout 和 return_when 若是設置了可能會返回未結束的future。

有一點你可能也注意到了，咱們重寫了get_flags 函數，是由於以前用到的 requests 庫執行的是阻塞型I/O操做。爲了使用 asyncio 包，咱們必須把函數改爲異步版。

小技巧

若是你以爲 使用了協程後代碼難以理解，能夠採用 Python之父（Guido van Rossum）的建議，僞裝沒有yield from。

已上邊這段代碼爲例：

@asyncio.coroutine
def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url) 
    image = yield from resp.read()
    return image

# 把yield form 去掉

def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = aiohttp.request('GET', url) 
    image = resp.read()
    return image

# 如今是否是清晰多了

知識點

在asyncio 包的API中使用 yield from 時，有個細節要注意：

使用asyncio包時，咱們編寫的異步代碼中包含由asyncio自己驅動的協程（委派生成器），而生成器最終把指責委託給asyncio包或者第三方庫中的協程。這種處理方式至關於架起了管道，讓asyncio事件循環驅動執行底層異步I/O的庫函數。

避免阻塞型調用

咱們先看一個圖，這個圖顯示了電腦從不一樣存儲介質中讀取數據的延遲狀況：

經過這個圖，咱們能夠看到，阻塞型調用對於CPU來講是巨大的浪費。有什麼辦法能夠避免阻塞型調用停止整個應用程序麼？

有兩種方法：

• 在單獨的線程中運行各個阻塞型操做

• 把每一個阻塞型操做轉化成非阻塞的異步調用使用

固然咱們推薦第二種方案，由於第一種方案中若是每一個鏈接都使用一個線程，成本過高。
第二種咱們可使用把生成器當作協程使用的方式實現異步編程。對事件循環來講，調用回調與在暫停的協程上調用 .send() 方法效果差很少。各個暫停的協程消耗的內存比線程小的多。

如今，你應該能理解爲何 flags_asyncio.py 腳本比 flags.py 快的多了吧。

由於flags.py 是依次同步下載，每次下載都要用幾十億個CPU週期等待結果。而在flags_asyncio.py中，在download_many 函數中調用loop.run_until_complete 方法時，事件循環驅動各個download_one 協程，運行到yield from 表達式出，那個表達式又驅動各個 get_flag 協程，運行到第一個yield from 表達式處，調用 aiohttp.request()函數。這些調用不會阻塞，所以在零點幾秒內全部請求均可以所有開始。

改進 asyncio 下載腳本

如今咱們改進一下上邊的 flags_asyncio.py，在其中添加上異常處理，計數器

import asyncio
import collections
from collections import namedtuple
from enum import Enum

import aiohttp
from aiohttp import web

from flags import save_flag, show, main, BASE_URL

DEFAULT_CONCUR_REQ = 5
MAX_CONCUR_REQ = 1000

Result = namedtuple('Result', 'status data')
HTTPStatus = Enum('Status', 'ok not_found error')

# 自定義異經常使用於包裝其餘HTTP貨網絡異常，並獲取country_code，以便報告錯誤
class FetchError(Exception):
    def __init__(self, country_code):
        self.country_code = country_code


@asyncio.coroutine
def get_flag(cc):
    # 此協程有三種返回結果：
    # 1.  返回下載到的圖片
    # 2. HTTP 響應爲404 時，拋出web.HTTPNotFound 異常
    # 3. 返回其餘HTTP狀態碼時， 拋出aiohttp.HttpProcessingError
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    if resp.status == 200:
        image = yield from resp.read()
        return image
    elif resp.status == 404:
        raise web.HttpNotFound()
    else:
        raise aiohttp.HttpProcessionError(
            code=resp.status, message=resp.reason,
            headers=resp.headers
        )


@asyncio.coroutine
def download_one(cc, semaphore):
    # semaphore 參數是 asyncio.Semaphore 類的實例
    # Semaphore 類是同步裝置，用於限制併發請求
    try:
        with (yield from semaphore):
             # 在yield    from  表達式中把semaphore   當成上下文管理器使用，防止阻塞整個系統
             # 若是semaphore 計數器的值是所容許的最大值，只有這個協程會阻塞
              image = yield from get_flag(cc)
              # 退出with語句後 semaphore 計數器的值會遞減，
              # 解除阻塞可能在等待同一個semaphore對象的其餘協程實例
    except web.HTTPNotFound:
        status = HTTPStatus.not_found
        msg = 'not found'
    except Exception as exc:
        raise FetchError(cc) from exc
    else:
        save_flag(image, cc.lower() + '.gif')
        status = HTTPStatus.ok
        msg = 'ok'
    return Result(status, cc)

@asyncio.coroutine
def downloader_coro(cc_list):
    counter = collections.Counter()
    # 建立一個 asyncio.Semaphore 實例，最多容許激活MAX_CONCUR_REQ個使用這個計數器的協程
    semaphore = asyncio.Semaphore(MAX_CONCUR_REQ)
    # 屢次調用 download_one 協程，建立一個協程對象列表
    to_do = [download_one(cc, semaphore) for cc in sorted(cc_list)]
    # 獲取一個迭代器，這個迭代器會在future運行結束後返回future
    to_do_iter = asyncio.as_completed(to_do)
    for future in to_do_iter:
        # 迭代容許結束的 future    
        try:
            res = yield from future # 獲取asyncio.Future 對象的結果（也能夠調用future.result）
        except FetchError as exc:
            # 拋出的異常都包裝在FetchError  對象裏
            country_code = exc.country_code
            try:
                # 嘗試從原來的異常 （__cause__）中獲取錯誤消息
                error_msg = exc.__cause__.args[0]
            except IndexError:
                # 若是在原來的異常中找不到錯誤消息，使用所鏈接異常的類名做爲錯誤消息
                error_msg = exc.__cause__.__class__.__name__
            if error_msg:
                msg = '*** Error for {}: {}'
                print(msg.format(country_code, error_msg))
            status = HTTPStatus.error
        else:
            status = res.status
        counter[status] += 1
    return counter

def download_many(cc_list):
    loop = asyncio.get_event_loop()
    coro = downloader_coro(cc_list)
    counts = loop.run_until_complete(coro)
    loop.close()
    return counts


if __name__ == '__main__':
    main(download_many)

因爲協程發起的請求速度較快，爲了防止向服務器發起太多的併發請求，使服務器過載，咱們在download_coro 函數中建立一個asyncio.Semaphore 實例，而後把它傳給download_one 函數。

<secion class="caption">Semaphore</section> 對象維護着一個內部計數器，若在對象上調用 .acquire() 協程方法，計數器則遞減；若在對象上調用 .release() 協程方法，計數器則遞增。計數器的值是在初始化的時候設定。
若是計數器大於0，那麼調用 .acquire() 方法不會阻塞，若是計數器爲0， .acquire() 方法會阻塞調用這個方法的協程，直到其餘協程在同一個 Semaphore 對象上調用 .release() 方法，讓計數器遞增。

在上邊的代碼中，咱們並無手動調用 .acquire() 或 .release() 方法，而是在 download_one 函數中 把 semaphore 當作上下文管理器使用:

with (yield from semaphore):
    image = yield from get_flag(cc)

這段代碼保證，任什麼時候候都不會有超過 MAX_CONCUR_REQ 個 get_flag 協程啓動。

使用 asyncio.as_completed 函數

由於要使用 yield from 獲取 asyncio.as_completed 函數產出的future的結果，因此 as_completed 函數秩序在協程中調用。因爲 download_many 要做爲參數傳給非協程的main 函數，我已咱們添加了一個新的 downloader_coro 協程，讓download_many 函數只用於設置事件循環。

使用Executor 對象，防止阻塞事件循環

如今咱們回去看下上邊關於電腦從不一樣存儲介質讀取數據的延遲狀況圖，有一個實時須要注意，那就是訪問本地文件系統也會阻塞。
上邊的代碼中，save_flag 函數阻塞了客戶代碼與 asyncio 事件循環公用的惟一線程，所以保存文件時，整個應用程序都會暫停。爲了不這個問題，可使用事件循環對象的 run_in_executor 方法。

asyncio 的事件循環在後臺維護着一個ThreadPoolExecutor 對象，咱們能夠調用 run_in_executor 方法，把可調用的對象發給它執行。
下邊是咱們改動後的代碼：

@asyncio.coroutine
def download_one(cc, semaphore):
    try:
        with (yield from semaphore):
            image = yield from get_flag(cc)
    except web.HTTPNotFound:
        status = HTTPStatus.not_found
        msg = 'not found'
    except Exception as exc:
        raise FetchError(cc) from exc
    else:
        # 這裏是改動部分
        loop = asyncio.get_event_loop()  # 獲取事件循環的引用
        loop.run_in_executor(None, save_flag, image, cc.lower() + '.gif')
        status = HTTPStatus.ok
        msg = 'ok'
    return Result(status, cc)

run_in_executor 方法的第一個參數是Executor 實例；若是設爲None,使用事件循環的默認 ThreadPoolExecutor 實例。

從回調到future到協程

在接觸協程以前，咱們可能對回調有必定的認識，那麼和回調相比，協程有什麼改進呢？

python中的回調代碼樣式：

def stage1(response1):
    request2 = step1(response1)
    api_call2(request2, stage2)
    
def stage2(response2):
    request3 = step3(response3)
    api_call3(request3, stage3)   

 def stage3(response3):
     step3(response3) 

api_call1(request1, stage1)

上邊的代碼的缺陷：

• 容易出現回調地獄

• 代碼難以閱讀

在這個問題上，協程能發揮很大的做用。若是換成協程和yield from 結果作的異步代碼，代碼示例以下：

@asyncio.coroutine
def three_stages(request1):
    response1 = yield from api_call1(request1)
    request2 = step1(response1)
    response2 = yield from api_call2(requests)
    request3 = step2(response2)
    response3 = yield from api_call3(requests)
    step3(response3)  
    
loop.create_task(three_stages(request1)

和以前的代碼相比，這個代碼就容易理解多了。若是異步調用 api_call1,api_call2,api_call3 會拋出異常，那麼能夠把相應的 yield from 表達式放在 try/except 塊中處理異常。
使用協程必須習慣 yield from 表達式，而且協程不能直接調用，必須顯式的排定協程的執行時間，或在其餘排定了執行時間的協程中使用yield from 表達式吧它激活。若是不使用 loop.create_task(three_stages(request1))，那麼什麼都不會發生。

下面咱們用一個實際的例子來演示一下：

每次下載發起屢次請求

咱們修改一下上邊下載國旗的代碼，使在下載國旗的同時還能夠獲取國家名稱在保存圖片的時候使用。
咱們使用協程和yield from 解決這個問題：

@asyncio.coroutine
def http_get(url):
    resp = yield from aiohttp.request('GET', url)
    if resp.status == 200:
        ctype = resp.headers.get('Content-type', '').lower()
        if 'json' in ctype or url.endswith('json'):
            data = yield from resp.json()
        else:
            data = yield from resp.read()
        return data
    elif resp.status == 404:
        raise web.HttpNotFound()
    else:
        raise aiohttp.HttpProcessionError(
            code=resp.status, message=resp.reason,
            headers=resp.headers)


@asyncio.coroutine
def get_country(cc):
    url = "{}/{cc}/metadata.json".format(BASE_URL, cc=cc.lower())
    metadata = yield from http_get(url)
    return metadata['country']


@asyncio.coroutine
def get_flag(cc):
    url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
    return (yield from http_get(url))


@asyncio.coroutine
def download_one(cc, semaphore):
    try:
        with (yield from semaphore):
            image = yield from get_flag(cc)
        with (yield from semaphore):
            country = yield from get_country(cc)
    except web.HTTPNotFound:
        status = HTTPStatus.not_found
        msg = 'not found'
    except Exception as exc:
        raise FetchError(cc) from exc
    else:
        country = country.replace(' ', '_')
        filename = '{}--{}.gif'.format(country, cc)
        print(filename)
        loop = asyncio.get_event_loop()
        loop.run_in_executor(None, save_flag, image, filename)
        status = HTTPStatus.ok
        msg = 'ok'
    return Result(status, cc)

在這段代碼中，咱們在download_one 函數中分別在 semaphore 控制的兩個with 塊中調用get_flag 和 get_country，是爲了節約時間。

get_flag 的return 語句在外層加上括號，是由於() 的運算符優先級高，會先執行括號內的yield from 語句 返回的結果。若是不加 會報句法錯誤
加() ，至關於

image = yield from http_get(url)
return image

若是不加()，那麼程序會在 yield from 處中斷，交出控制權，這時使用return 會報句法錯誤。

總結

這一篇咱們討論了：

• 對比了一個多線程程序和asyncio版，說明了多線程和異步任務之間的關係

• 比較了 asyncio.Future 類 和 concurrent.futures.Future 類的區別

• 如何使用異步編程管理網絡應用中的高併發

• 在異步編程中，與回調相比，協程顯著提高性能的方式

下一篇，咱們將介紹如何使用asyncio包編寫服務器

參考連接

• class asyncio.Semaphore

• asyncio — Asynchronous I/O, event loop, coroutines and tasks

• 【譯】 Python 3.5 協程到底是個啥

• PEP 0492 Coroutines with async and await syntax

• Python 之 asyncio

• 我所不能理解的Python中的Asyncio模塊

最後，感謝女友支持

>歡迎關注	>請我喝芬達