[譯]Python中的異步IO:一個完整的演練

時間 2020-05-07

原文原文鏈接

原文：Async IO in Python: A Complete Walkthrough 原文做者： Brad Solomon 原文發佈時間：2019年1月16日翻譯：Tacey Wong 翻譯時間：2019年7月22日javascript

翻譯僅便於我的學習，熟悉英語的請閱讀原文html

目錄java

搭建本身的實驗環境
異步IO鳥瞰圖
asyncio 包和 async/await
- async/await 語法和原生協程
- 異步IO規則
異步IO設計模式
- 鏈式協程
- 使用隊列
生成器中異步IO的Roots
- 其餘特色: async for and Async Generators + Comprehensions
- 事件循環和asyncio.run()
一個完整的程序：異步請求
上下文中的異步IO
- 什麼時候以及爲什麼異步IO是正確的選擇？
- Async IO It Is, but Which One?
其餘零碎
- [其餘頂級asyncio 函數](#其餘頂級asyncio 函數)
- await的優先級
總結
附加資源

Async IO是一種併發編程設計，Python中已經有了獨立的支持，而且從Python3.4到Python3.7獲得了快速發展。node

你可能疑惑，「併發、並行、線程、多處理」。MMP這已經不少了，異步IO是哪根蔥？」python

本教程旨在幫助你回答這個問題，讓你更牢固地掌握Python的異步IO。ios

如下是要介紹的內容：git

異步IO：一種與語言無關的範例(模型)，它具備許多跨編程語言的實現
async/await:兩個用於定義協程的新Python關鍵字
asyncio：爲運行和管理協程提供基礎和API的Python包/庫

協程（專用生成器函數）是Python中異步IO的核心，稍後咱們將深刻研究它們。程序員

注意：在本文中，使用術語異步IO來表示與語言無關的異步IO設計，而asyncio指的是Python包。github

開始以前，你須要確保已經配置搭建了可使用asyncio及其餘庫的實驗環境。golang

搭建本身的實驗環境

你須要安裝Python 3.7+以及aiohttp和aiofiles包才能完整地跟隨本文進行實驗。

$ python3.7 -m venv ./py37async
$ source ./py37async/bin/activate  # Windows: .\py37async\Scripts\activate.bat
$ pip install --upgrade pip aiohttp aiofiles  # 可選項: aiodns

有關安裝Python 3.7和設置虛擬環境的幫助，請查看Python 3安裝和設置指南或虛擬環境基礎

ok,let's go!

異步IO鳥瞰圖

相較於它久經考驗的表親(多進程和多線程)來講，異步IO不太爲人所知。本節將從高層全面地介紹異步IO是什麼，以及哪些場景適合用它。

哪些場景適合異步IO？

併發和並行是個很是普遍的主題。由於本文重點介紹異步IO及其在Python中的實現，如今值得花一點時間將異步IO與其對應物進行比較，以瞭解異步IO如何適應更大、有時使人眼花繚亂的難題。

並行：同時執行多個操做。 多進程：是一種實現並行的方法，它須要將任務分散到計算機的中央處理單元(cpu或核心)上。多進程很是適合cpu密集的任務:密集for循環和密集數學計算一般屬於這一類。併發：併發是一個比並行更普遍的術語。它代表多個任務可以以重疊方式運行。（有一種說法是併發並不意味着並行。）線程：是一種併發執行模型，多個線程輪流執行任務。一個進程能夠包含多個線程。因爲GIL（全局解釋器鎖）的存在，Python與線程有着複雜的關係，但這超出了本文的範圍。

瞭解線程的重要之處是它更適合於io密集的任務。cpu密集型任務的特色是計算機核心從開始到結束都在不斷地工做，而一個IO密集型任務更多的是等待IO的完成。

綜上所述，併發既包括多進程(對於CPU密集任務來講是理想的)，也包括線程(對於IO密集型任務來講是理想的)。多進程是並行的一種形式，並行是併發的一種特定類型(子集)。Python經過multiprocessing, threading, 和concurrent.futures標準庫爲這二者提供了長期支持。

如今是時候召集一名新成員了！在過去的幾年裏，一個獨立的設計被更全面地嵌入到了CPython中：經過標準庫的asyncio包和新的async/await語言關鍵字實現異步IO。須要說明的是，異步IO不是一個新發明的概念，它已經存在或正在構建到其餘語言和運行時環境中，好比Golang、C#或者Scala。

Python文檔將asyncio包稱爲用於編寫併發代碼的庫。然而，異步IO既不是多線程也不是多進程，它不是創建在其中任何一個之上。事實上異步IO是一種單進程單線程設計：它使用協做式多任務操做方式，在本教程結束時你將理解這個術語。換句話說，儘管在單個進程中使用單個線程，但異步IO給人一種併發的感受。協程(異步IO的一個核心特性)能夠併發地調度，但它們本質上不是併發的。

重申一下，異步輸入輸出是併發編程的一種風格，但不是並行的。與多進程相比，它與線程更緊密地結合在一塊兒，但與這二者大相徑庭，而且是併發技術包中的獨立成員。

如今還留下了一個詞沒有解釋。異步是什麼意思？這不是一個嚴格的定義，可是對於咱們這裏的目的，我能夠想到/考慮到兩個屬性:

異步例程可以在等待其最終結果時「暫停」，並容許其餘例程同時運行。
經過上面的機制，異步代碼便於併發執行。換句話說，異步代碼提供了併發的外觀和感受

下面是一個一個將全部內容組合在一塊兒的圖表。白色術語表明概念，綠色術語表明實現或實現它們的方式：

（Concurrencey併發、Threading線程、Async IO異步IO、Parallelism並行、Multiprocessing多進程）

我將在這裏中止對併發編程模型的比較。本教程重點介紹異步IO的子組件，如何使用它、以及圍繞它建立的API。要深刻研究線程、多處理和異步IO，請暫停這裏並查看Jim Anderson對(Python中併發性的概述)[https://realpython.com/python-concurrency/]。Jim比我有趣得多，並且參加的會議也比我多。

譯者注：要了解多種併發模型的比較，能夠參考(《七週七併發模型》）

異步IO釋義

異步IO乍一看彷佛違反直覺，自相矛盾。如何使用一個線程和一個CPU內核來簡化併發代碼?我歷來都不擅長編造例子，因此我想借用 Miguel Grinberg2017年PyCon演講中的一個例子，這個例子很好地解釋了一切:

國際象棋大師JuditPolgár舉辦了一個國際象棋比賽，在那裏她扮演多個業餘選手。她有兩種方式進行比賽：同步和異步。假設：

24個對手
Judit在5秒鐘內完成一個棋子的移動
每一個對手移動一個棋子須要55秒
遊戲平均30對移動（總計60次移動） 同步版本：Judit一次只玩一場遊戲，從不一樣時玩兩場，直到遊戲結束。每場比賽須要（55 + 5）* 30 == 1800秒，或30分鐘。整個比賽須要24 * 30 == 720分鐘，或12小時。 異步版本：Judit從一張桌子走到另外一張桌子，每張桌子走一步。她離開了牌桌，讓對手在等待的時間裏採起下一步行動。在全部24場比賽中，一個動做須要Judit 24 * 5 == 120秒，即2分鐘。整個比賽如今被縮減到120 * 30 == 3600秒，也就是1小時。

只有一個JuditPolgár，她只有兩隻手，一次只作一次動做。可是，異步進行將展覽時間從12小時減小到1小時。所以，協同多任務處理是一種奇特的方式，能夠說一個程序的事件循環(稍後會有更多)與多個任務通訊，讓每一個任務在最佳時間輪流運行。

異步IO須要很長的等待時間，不然函數將被阻塞，並容許其餘函數在停機期間運行。

異步IO使用起來不容易

我聽人說過「當你可以的時候使用異步IO;必要時使用線程」。事實是，構建持久的多線程代碼可能很難，而且容易出錯。異步IO避免了一些線程設計可能遇到的潛在速度障礙。

但這並非說Python中的異步IO很容易。警告:當你稍微深刻其中時，異步編程也會很困難！Python的異步模型是圍繞諸如回調，事件，傳輸，協議和future等概念構建的 -——術語可能使人生畏。事實上，它的API一直在不斷變化，這使得它變得比較難。

幸運的是，asyncio已經相對成熟，其大部分功能再也不處於臨時性狀態，而其文檔也有了大規模的改善，而且該主題的一些優質資源也開始出現。

`asyncio` 包和 `async/await`

如今你已經對異步輸IO做爲一種設計有了必定的瞭解，讓咱們來探討一下Python的實現。Python的asyncio包(在Python 3.4中引入)和它的兩個關鍵字async和wait服務於不一樣的目的，可是它們會一塊兒幫助你聲明、構建、執行和管理異步代碼。

`async/await` 語法和原生協程

警告：當心你在網上讀到的東西。Python的異步IO API已經從Python 3.4迅速發展到Python 3.7。一些舊的模式再也不被使用，一些最初不被容許的東西如今經過新的引入被容許。據我所知，本教程也將很快加入過期的行列。

異步IO的核心是協程。協程是Python生成器函數的一個專門版本。讓咱們從一個基線定義開始，而後隨着你在此處的進展，以此爲基礎進行構建:協程是一個函數，它能夠在到達返回以前暫停執行，而且能夠在一段時間內間接將控制權傳遞給另外一個協程。

稍後，你將更深刻地研究如何將傳統生成器從新用於協程。目前，瞭解協程如何工做的最簡單方法是開始編寫一些協程代碼。

讓咱們採用沉浸式方法，編寫一些異步輸入輸出代碼。這個簡短的程序是異步IO的Hello World，但它對展現其核心功能大有幫助:

#!/usr/bin/env python3
# countasync.py

import asyncio

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    import time
    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

當你執行此文件時，請注意與僅用def和time.sleep()定義函數相比，看起來有什麼不一樣:

$ python3 countasync.py
One
One
One
Two
Two
Two
countasync.py executed in 1.01 seconds.

該輸出的順序是異步IO的核心。與count()的每一個調用通訊是一個事件循環或協調器。當每一個任務到達asyncio.sleep(1)時，函數會向事件循環發出呼叫，並將控制權交還給它，例如，「我將休眠1秒。在這段時間裏，作一些有意義的事情吧」。

將此與同步版本進行對比：：

#!/usr/bin/env python3
# countsync.py

import time

def count():
    print("One")
    time.sleep(1)
    print("Two")

def main():
    for _ in range(3):
        count()

if __name__ == "__main__":
    s = time.perf_counter()
    main()
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

執行時，順序和執行時間會有輕微但嚴重的變化：

$ python3 countsync.py
One
Two
One
Two
One
Two
countsync.py executed in 3.01 seconds.

雖然使用time.sleep()和asyncio.sleep()看起來很普通，可是它們能夠替代任何涉及等待時間的時間密集型進程。(您能夠等待的最普通的事情是一個sleep()調用，它基本上什麼也不作。)也就是說，time.sleep()能夠表示任何耗時的阻塞函數調用，而asyncio.sleep()用於代替非阻塞調用(但也須要一些時間來完成)。

你將在下一節中看到，等待某些東西(包括asyncio.sleep()的好處是，周圍的函數能夠暫時將控制權交給另外一個更容易當即執行某些操做的函數。相比之下，time.sleep()或任何其餘阻塞調用與異步Python代碼不兼容，由於它會在睡眠時間內中止全部工做。

異步IO規則

此時，異步、wait和它們建立的協程函數的更正式定義已經就緒。這一節有點密集，可是掌握async/await是頗有幫助的，因此若是須要的話，能夠回到這裏：

語法async def引入了原生協程或異步生成器。async with和async for表達式也是有效的，稍後你將看到它們。
關鍵詞await將函數控制傳遞迴事件循環（它暫停執行周圍的協程）。若是Python在g()的範圍內遇到await f()表達式，這就是await告訴事件循環，「暫停執行g()直到我等待的f()的結果返回。與此同時，讓其餘東西運行。「

在代碼中，第二個要點大體是這樣的:

async def g():
    # 在這裏暫停 ，f()執行完以後再返回到這裏。
    return r

關於什麼時候以及可否使用async / await，還有一套嚴格的規則。不管您是在學習語法仍是已經使用async / await，這些都很是方便：

使用async def引入的函數是協程。它可使用wait、return或yield，但全部這些都是可選的。聲明async def noop(): pass 是合法的：
- 使用wait和/或return建立一個coroutine函數。要調用coroutine函數，你必須等待它獲得結果。
- 在異步def塊中使用yield不太常見（而且最近纔在Python中合法）。這將建立一個異步生成器，您可使用異步生成器進行迭代。暫時忘掉異步生成器，重點關注使用await和/或return的協程函數的語法。
- 任何使用async def定義的東西都不能使用yield from，這會引起SyntaxError(語法錯誤)。
就像在def函數以外使用yield是一個SyntaxError同樣，在async def協程以外使用wait也是一個SyntaxError。

如下是一些簡潔的示例，旨在總結以上幾條規則：

async def f(x):
    y = await z(x)  # OK - `await` and `return` allowed in coroutines
    return y

async def g(x):
    yield x  # OK - this is an async generator

async def m(x):
    yield from gen(x)  # No - SyntaxError

def m(x):
    y = await z(x)  # Still no - SyntaxError (no `async def` here)
    return y

最後，當您使用await f()時，它要求f()是一個awaitable對象。嗯，這不是頗有幫助，是嗎？如今，只要知道一個等待對象是（1）另外一個協程或（2）定義返回一個迭代器.__ await __（）dunder方法的對象。若是你正在編寫一個程序，在大多數狀況下，你只須要擔憂第一種狀況。

這又給咱們帶來了一個你可能會看到的技術上的區別:將函數標記爲coroutine的一個老方法是用@asyncio.coroutine來修飾一個普通的def函數。結果是基於生成器的協同程序。自從在Python 3.5中引入async/await語法以來，這種結構已通過時了。

這兩個協程本質上是等價的(都是可 awaitable的)，可是第一個協程是基於生成器的，而第二個協程是一個原生協程：

import asyncio

@asyncio.coroutine
def py34_coro():
    """Generator-based coroutine, older syntax"""
    yield from stuff()

async def py35_coro():
    """Native coroutine, modern syntax"""
    await stuff()

若是你本身編寫任何代碼，爲了顯式最好使用本機協程。基於生成器的協程將在Python 3.10中刪除。

在本教程的後半部分，咱們將僅出於解釋的目的來討論基於生成器的協同程序。引入async / await的緣由是使協同程序成爲Python的獨立功能，能夠很容易地與正常的生成器函數區分開來，從而減小歧義。

不要陷入基於生成器的協程中，這些協同程序已隨着async / await的出現而過期了。若是你堅持async/await語法，它們有本身的小規則集(例如，await不能在基於生成器的協同程序中使用)，這些規則在很大程度上是不相關的。

廢話很少說，讓咱們來看幾個更復雜的例子。

下面是異步IO如何減小等待時間的一個例子：給定一個協程makerandom()，它一直在[0,10]範圍內產生隨機整數，直到其中一個超過閾值，你想讓這個協程的屢次調用不須要等待彼此連續完成。你能夠在很大程度上遵循上面兩個腳本的模式，只需稍做修改：

#!/usr/bin/env python3
# rand.py

import asyncio
import random

# ANSI colors
c = (
    "\033[0m",   # End of color
    "\033[36m",  # Cyan
    "\033[91m",  # Red
    "\033[35m",  # Magenta
)

async def makerandom(idx: int, threshold: int = 6) -> int:
    print(c[idx + 1] + f"Initiated makerandom({idx}).")
    i = random.randint(0, 10)
    while i <= threshold:
        print(c[idx + 1] + f"makerandom({idx}) == {i} too low; retrying.")
        await asyncio.sleep(idx + 1)
        i = random.randint(0, 10)
    print(c[idx + 1] + f"---> Finished: makerandom({idx}) == {i}" + c[0])
    return i

async def main():
    res = await asyncio.gather(*(makerandom(i, 10 - i - 1) for i in range(3)))
    return res

if __name__ == "__main__":
    random.seed(444)
    r1, r2, r3 = asyncio.run(main())
    print()
    print(f"r1: {r1}, r2: {r2}, r3: {r3}")

彩色輸出比我能說的多得多，並讓你瞭解這個腳本是如何執行的：

該程序使用一個主協程makerandom（），並在3個不一樣的輸入上同時運行它。大多數程序將包含小型、模塊化的協程和一個包裝器函數，用於將每一個較小的協程連接在一塊兒。而後，main()用中央協程映射到某個可迭代的池中收集任務(future)。

在這個小例子中，池是range(3)。在稍後介紹的更全面的示例中，它是一組須要同時請求，解析和處理的URL，main()封裝了每一個URL的整個例程。

雖然「製做隨機整數」（CPU密集比這更復雜）可能不是做爲asyncio候選者的最佳選擇，可是在示例中存在asyncio.sleep（），旨在模仿不肯定等待時間的IO密集進程。例如，asyncio.sleep（）調用可能表示在消息應用程序中的兩個客戶端之間發送和接收不那麼隨機的整數。

異步IO設計模式

Async IO附帶了它本身的一組腳本設計，您將在本節中介紹這些腳本設計。

鏈式協程

協程的一個關鍵特性是它們能夠連接在一塊兒（記住，一個協成對象是awaitable的，因此另一個協成能夠await它）。這容許你將程序分紅更小的、可管理的、可回收的協同程序:

#!/usr/bin/env python3
# chained.py

import asyncio
import random
import time

async def part1(n: int) -> str:
    i = random.randint(0, 10)
    print(f"part1({n}) sleeping for {i} seconds.")
    await asyncio.sleep(i)
    result = f"result{n}-1"
    print(f"Returning part1({n}) == {result}.")
    return result

async def part2(n: int, arg: str) -> str:
    i = random.randint(0, 10)
    print(f"part2{n, arg} sleeping for {i} seconds.")
    await asyncio.sleep(i)
    result = f"result{n}-2 derived from {arg}"
    print(f"Returning part2{n, arg} == {result}.")
    return result

async def chain(n: int) -> None:
    start = time.perf_counter()
    p1 = await part1(n)
    p2 = await part2(n, p1)
    end = time.perf_counter() - start
    print(f"-->Chained result{n} => {p2} (took {end:0.2f} seconds).")

async def main(*args):
    await asyncio.gather(*(chain(n) for n in args))

if __name__ == "__main__":
    import sys
    random.seed(444)
    args = [1, 2, 3] if len(sys.argv) == 1 else map(int, sys.argv[1:])
    start = time.perf_counter()
    asyncio.run(main(*args))
    end = time.perf_counter() - start
    print(f"Program finished in {end:0.2f} seconds.")

請仔細注意輸出，其中part1()睡眠時間可變，part2()在結果可用時開始處理結果:

$ python3 chained.py 9 6 3
part1(9) sleeping for 4 seconds.
part1(6) sleeping for 4 seconds.
part1(3) sleeping for 0 seconds.
Returning part1(3) == result3-1.
part2(3, 'result3-1') sleeping for 4 seconds.
Returning part1(9) == result9-1.
part2(9, 'result9-1') sleeping for 7 seconds.
Returning part1(6) == result6-1.
part2(6, 'result6-1') sleeping for 4 seconds.
Returning part2(3, 'result3-1') == result3-2 derived from result3-1.
-->Chained result3 => result3-2 derived from result3-1 (took 4.00 seconds).
Returning part2(6, 'result6-1') == result6-2 derived from result6-1.
-->Chained result6 => result6-2 derived from result6-1 (took 8.01 seconds).
Returning part2(9, 'result9-1') == result9-2 derived from result9-1.
-->Chained result9 => result9-2 derived from result9-1 (took 11.01 seconds).
Program finished in 11.01 seconds.

在此設置中，main()的運行時間將等於它收集和調度的任務的最大運行時間。

使用隊列

asyncio包提供了與queue模塊的類相似的queue classes類。在咱們到目前爲止的示例中，咱們並不真正須要隊列結構。在 chained.py中,每一個任務(future)都由一組協同程序組成，這些協同程序顯式地相互等待，並在每一個鏈上傳遞一個輸入。

還有一種替代結構也能夠用於異步IO：許多生產者，彼此沒有關聯，將項目添加到隊列中。每一個生產者能夠在交錯、隨機、未宣佈的時間向隊列添加多個項。當商品出現時，一組消費者貪婪地從隊列中取出商品，不等待任何其餘信號。

在這種設計中，沒有任何個體消費者與生產者的連接。消費者事先不知道生產者的數量，甚至不知道將添加到隊列中的累計項目數。

單個生產者或消費者分別從隊列中放置和提取項所需的時間是可變的。隊列充當一個吞吐量，它能夠與生產者和消費者通訊，而不須要它們彼此直接通訊。

注意：雖然隊列一般用於線程程序，由於queue.Queue()的線程安全性。在涉及異步IO時，您不須要關心線程安全性(例外狀況是當你將二者結合時，但在本教程中沒有這樣作。)【譯者注：這裏的二者結合說的是異步IO和多線程結合】。隊列的一個用例(如這裏的例子)是隊列充當生產者和消費者的發送器，不然它們不會直接連接或關聯在一塊兒。

這個程序的同步版本看起來至關糟糕：一組阻塞生成器按順序將項添加到隊列中，一次一個生產者。只有在全部生產者完成以後，隊列才能夠由一個消費者一次處理一個項一個項地處理。這種設計有大量的延遲。物品可能會閒置在隊列中，而不是當即拿起並處理。

下面是異步版本asyncq.py。這個工做流程的挑戰在於須要向消費者發出生產完成的信號。不然，await q.get()將無限期掛起，由於隊列已經被徹底處理，可是消費者並不知道生產已經完成。

(很是感謝StackOverflow用戶幫助理順main():關鍵是await q.join（），它將一直阻塞到隊列中的全部項都被接收和處理，而後取消消費者任務，不然這些任務會掛起並沒有休止地等待其餘隊列項出現)

下面是完整的腳本：

#!/usr/bin/env python3
# asyncq.py

import asyncio
import itertools as it
import os
import random
import time

async def makeitem(size: int = 5) -> str:
    return os.urandom(size).hex()

async def randsleep(a: int = 1, b: int = 5, caller=None) -> None:
    i = random.randint(0, 10)
    if caller:
        print(f"{caller} sleeping for {i} seconds.")
    await asyncio.sleep(i)

async def produce(name: int, q: asyncio.Queue) -> None:
    n = random.randint(0, 10)
    for _ in it.repeat(None, n):  # Synchronous loop for each single producer
        await randsleep(caller=f"Producer {name}")
        i = await makeitem()
        t = time.perf_counter()
        await q.put((i, t))
        print(f"Producer {name} added <{i}> to queue.")

async def consume(name: int, q: asyncio.Queue) -> None:
    while True:
        await randsleep(caller=f"Consumer {name}")
        i, t = await q.get()
        now = time.perf_counter()
        print(f"Consumer {name} got element <{i}>"
              f" in {now-t:0.5f} seconds.")
        q.task_done()

async def main(nprod: int, ncon: int):
    q = asyncio.Queue()
    producers = [asyncio.create_task(produce(n, q)) for n in range(nprod)]
    consumers = [asyncio.create_task(consume(n, q)) for n in range(ncon)]
    await asyncio.gather(*producers)
    await q.join()  # Implicitly awaits consumers, too
    for c in consumers:
        c.cancel()

if __name__ == "__main__":
    import argparse
    random.seed(444)
    parser = argparse.ArgumentParser()
    parser.add_argument("-p", "--nprod", type=int, default=5)
    parser.add_argument("-c", "--ncon", type=int, default=10)
    ns = parser.parse_args()
    start = time.perf_counter()
    asyncio.run(main(**ns.__dict__))
    elapsed = time.perf_counter() - start
    print(f"Program completed in {elapsed:0.5f} seconds.")

前幾個協同程序是輔助函數，它返回隨機字符串，小數秒性能計數器和隨機整數。生產者將1到5個項目放入隊列中。每一個項目是（i，t）的元組，其中i是隨機字符串，t是生產者嘗試將元組放入隊列的時間。

當消費者將項目拉出時，它只使用項目所在的時間戳計算項目在隊列中所用的時間。

請記住，asyncio.sleep()是用來模擬其餘一些更復雜的協同程序的，若是它是一個常規的阻塞函數，會消耗時間並阻塞全部其餘的執行。 .

下面是一個有兩個生產者和五個消費者的測試：

$ python3 asyncq.py -p 2 -c 5
Producer 0 sleeping for 3 seconds.
Producer 1 sleeping for 3 seconds.
Consumer 0 sleeping for 4 seconds.
Consumer 1 sleeping for 3 seconds.
Consumer 2 sleeping for 3 seconds.
Consumer 3 sleeping for 5 seconds.
Consumer 4 sleeping for 4 seconds.
Producer 0 added <377b1e8f82> to queue.
Producer 0 sleeping for 5 seconds.
Producer 1 added <413b8802f8> to queue.
Consumer 1 got element <377b1e8f82> in 0.00013 seconds.
Consumer 1 sleeping for 3 seconds.
Consumer 2 got element <413b8802f8> in 0.00009 seconds.
Consumer 2 sleeping for 4 seconds.
Producer 0 added <06c055b3ab> to queue.
Producer 0 sleeping for 1 seconds.
Consumer 0 got element <06c055b3ab> in 0.00021 seconds.
Consumer 0 sleeping for 4 seconds.
Producer 0 added <17a8613276> to queue.
Consumer 4 got element <17a8613276> in 0.00022 seconds.
Consumer 4 sleeping for 5 seconds.
Program completed in 9.00954 seconds.

在這種狀況下，項目在幾分之一秒內處理。延遲可能有兩個緣由：

標準的，在很大程度上不可避免的開銷
當一個項出如今隊列中時，全部消費者都在睡覺的狀況

關於第二個緣由，幸運的是，擴展到成百上千的消費者是徹底正常的。用python3 asyncq.py -p 5 - c100應該沒有問題。這裏的要點是，理論上，您可讓不一樣系統上的不一樣用戶控制生產者和消費者的管理，隊列充當中央吞吐量。

到目前爲止，您已經跳進了火坑。瞭解了三個asyncio調用async和await定義的協程並等待的示例。若是你沒有徹底關注或者只是想深刻了解Python中現代協同程序的機制，下一節咱們將開始討論這個。

生成器中異步IO的Roots

以前，您看到了一個基於生成器的舊式協同程序的例子，它已經被更顯式的原生協同程序所淘汰。這個例子值得從新展現一下：

import asyncio

@asyncio.coroutine
def py34_coro():
    """Generator-based coroutine"""
    # No need to build these yourself, but be aware of what they are
    s = yield from stuff()
    return s

async def py35_coro():
    """Native coroutine, modern syntax"""
    s = await stuff()
    return s

async def stuff():
    return 0x10, 0x20, 0x30

做一個實驗，若是py34_coro()或py35_coro()調用自身，而不await或不調用asyncio.run()或其餘asyncio函數，會發生什麼？獨調用一個協同程序會返回一個協同程序對象：

>>> py35_coro()
<coroutine object py35_coro at 0x10126dcc8>

這表面上並非頗有趣。調用協同程序的結果是一個awaitable的協程對象。

測驗時間：Python的其餘什麼功能跟這同樣？(Python的哪些特性在單獨調用時實際上沒有多大做用?)

但願你將生成器做爲這個問題的答案，由於協同程序是加強型生成器。在這方面的行爲相似：

>>> def gen():
...     yield 0x10, 0x20, 0x30
...
>>> g = gen()
>>> g  # Nothing much happens - need to iterate with `.__next__()`
<generator object gen at 0x1012705e8>
>>> next(g)
(16, 32, 48)

正如它所發生的那樣，生成器函數是異步IO的基礎（不管是否使用async def聲明協程而不是舊的@asyncio.coroutine包裝器）。從技術上講，await更接近於yield from而非yield。（但請記住，yield from x()只是替換for i in x():yield i的語法糖）

生成器與異步IO相關的一個關鍵特性是能夠有效地隨意中止和從新啓動生成器。例如，你能夠在生成器對象上進行迭代，而後在剩餘的值上繼續迭代。當一個生成器函數達到yield時，它會產生該值，但隨後它會處於空閒狀態，直到它被告知產生其後續值。

這能夠經過一個例子來充實:

>>> from itertools import cycle
>>> def endless():
...     """Yields 9, 8, 7, 6, 9, 8, 7, 6, ... forever"""
...     yield from cycle((9, 8, 7, 6))

>>> e = endless()
>>> total = 0
>>> for i in e:
...     if total < 30:
...         print(i, end=" ")
...         total += i
...     else:
...         print()
...         # Pause execution. We can resume later.
...         break
9 8 7 6 9 8 7 6 9 8 7 6 9 8

>>> # Resume
>>> next(e), next(e), next(e)
(6, 9, 8)

await關鍵字的行爲相似，標記了一個斷點，協程掛起本身並容許其餘協程工做。在這種狀況下，「掛起」是指暫時放棄控制但未徹底退出或結束協程。請記住，yield，以及由此產生的yield from和await是發生器執行過程當中的一個斷點。

這是函數和生成器之間的根本區別。一個函數要麼全有要麼全無。一旦它開始，它就不會中止，直到它到達一個return，而後將該值推給調用者(調用它的函數)。另外一方面，生成器每次達到yield時都會暫停，再也不繼續。它不只能夠將這個值推入調用堆棧，並且當您經過對它調用next()恢復它時，它還能夠保留它的局部變量。

生成器的第二個特徵雖然不爲人知，卻也也很重要。也能夠經過其.send（）方法將值發送到生成器。這容許生成器（和協同程序）相互調用（await）而不會阻塞。我不會再深刻了解這個功能的細節，由於它主要是爲了在幕後實現協同程序，但你不該該真的須要本身直接使用它。

若是你有興趣瞭解更多內容，能夠從PEP 342/正式引入協同程序開始。 Brett Cannon的Python中異步等待(Async-Await)是如何工做的也是一個很好的讀物,asyncio上的PYMOTW文章也是如此。還有David Beazley的[關於協程和併發的有趣課程] 深刻探討了協同程序運行的機制。

讓咱們嘗試將上述全部文章壓縮成幾句話：

這些協同程序其實是經過一種很是規的機制運行的。它們的結果是在調用其.send（）方法時拋出異常對象的屬性。全部這些都有一些不可靠的細節，可是它可能不會幫助您在實踐中使用這部分語言，因此如今讓咱們繼續。

爲了聯繫在一塊兒，如下是關於協同做爲生成器這個主題的一些關鍵點：

協同程序是利用生成器方法的特性的再利用生成器。
舊式基於生成器的協同程序使用yield from來等待協程結果。原生協同程序中的現代Python語法只是將yield from等價替換爲await做爲等待協程結果的方法。await相似於yield，這樣想一般是有幫助的。
await的使用是標誌着斷點的信號。它容許協程暫時暫停執行並容許程序稍後返回它。

其餘特色: async for and Async Generators + Comprehensions

與純async/await一塊兒，Python還容許經過async for異步迭代異步迭代器。異步迭代器的目的是讓它可以在迭代時在每一個階段調用異步代碼。

這個概念的天然延伸是異步發生器。回想一下，你能夠在原生協程中使用await，return或yield。在Python 3.6中可使用協程中的yield（經過PEP 525），它引入了異步生成器，目的是容許await和yield在同一個協程函數體中使用：

>>> async def mygen(u: int = 10):
...     """Yield powers of 2."""
...     i = 0
...     while i < u:
...         yield 2 ** i
...         i += 1
...         await asyncio.sleep(0.1)

最後但一樣重要的是，Python經過async for來實現異步理解。就像它的同步表兄弟同樣，這主要是語法糖：

>>> async def main():
...     # This does *not* introduce concurrent execution
...     # It is meant to show syntax only
...     g = [i async for i in mygen()]
...     f = [j async for j in mygen() if not (j // 3 % 5)]
...     return g, f
...
>>> g, f = asyncio.run(main())
>>> g
[1, 2, 4, 8, 16, 32, 64, 128, 256, 512]
>>> f
[1, 2, 16, 32, 256, 512]

這是一個關鍵的區別：異步生成器和理解都不會使迭代併發。它們所作的就是提供同步對等程序的外觀和感受，可是有能力讓循環放棄對事件循環的控制，讓其餘協同程序運行。

換句話說，異步迭代器和異步生成器不是爲了在序列或迭代器上同時映射某些函數而設計的。它們僅僅是爲了讓封閉的協程容許其餘任務輪流使用。async for和async with語句僅在使用純for或with會「破壞」協程中await的性質的狀況下才須要。異步性和併發之間的區別是一個須要掌握的關鍵因素。

事件循環和`asyncio.run()`

您能夠將事件循環視爲一段時間的while True循環，它監視協同程序，獲取有關閒置內容的反饋，並查找可在此期間執行的內容。當協同程序等待的任何內容變得可用時，它可以喚醒空閒協程。

到目前爲止，事件循環的整個管理已由一個函數調用隱式處理：

asyncio.run(main())  # Python 3.7+

Python 3.7中引入的asyncio.run()負責獲取事件循環，運行任務直到它們被標記爲完成，而後關閉事件循環。

使用get_event_loop（）管理asyncio事件循環有一種更加冗長的方式。典型的模式以下所示:

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

你可能會在較舊的示例中看到loop.get_event_loop（），但除非你須要對事件循環管理控制進行特別的微調，不然asyncio.run（）應該足以知足大多數程序的須要。

若是確實須要在Python程序中與事件循環交互，loop是一個老式的Python對象，它支持使用loop.is_running（）和loop.is_closed（）進行內省/introspection 。若是須要得到更精細的控制，能夠對其進行操做，例如經過將循環做爲參數傳遞來調度回調。

更重要的是要深刻了解事件循環的機制。關於事件循環，這裏有幾點值得強調。

1：協同程序在與事件循環綁定以前不會自行作不少事情。

你以前在生成器的解釋中看到了這一點，但值得重申。若是您有一個主協程在等待其餘協程，那麼單獨調用它幾乎沒有什麼效果：

>>> import asyncio

>>> async def main():
...     print("Hello ...")
...     await asyncio.sleep(1)
...     print("World!")

>>> routine = main()
>>> routine
<coroutine object main at 0x1027a6150>

請記住使用asyncio.run（）經過調度main（）協程（未來的對象）來實際強制執行，以便在事件循環上執行：

>>> asyncio.run(routine)
Hello ...
World!

(其餘協同程序能夠經過await執行。一般在asyncio.run()中封裝main()，而後從那裏調用帶有await的鏈式協程。）

2：默認狀況下，異步IO事件循環在單個線程和單個CPU內核上運行。一般，在一個CPU內核中運行一個單線程事件循環是綽綽有餘的。還能夠跨多個核心運行事件循環。請查看John Reese談話獲取更多內容，順便提個醒，你的筆記本電腦可能會自發燃燒。

3：事件循環是可插入的。也就是說，若是你真的須要，你能夠編寫本身的事件循環實現，並讓它以相同的方式運行任務。這在uvloop包中獲得了很好的演示，這是Cython中事件循環的一個實現。

這就是"可插入事件循環"這個術語的含義:你可使用事件循環的任何工做實現，與協同程序自己的結構無關。asyncio包自己附帶兩個不一樣的事件循環實現，默認狀況下基於選擇器模塊。(第二個實現僅適用於Windows。)

一個完整的程序：異步請求

你已經走了這麼遠，如今是時候享受快樂和無痛的部分了。在本節中，您將使用aiohttp(一種速度極快的異步http 客戶端/服務端框架)構建一個抓取網頁的網址收集器areq.py。（咱們只須要客戶端部分。）這種工具能夠用來映射一組站點之間的鏈接，這些連接造成一個有向圖。

注：您可能想知道爲何Python的requests包與異步IO不兼容。requests構建在urllib3之上，而urllib3又使用Python的http和socket模塊。默認狀況下，socket操做是阻塞的。這意味着Python不會想await requests.get（url）這樣，由於.get（）不是awaitable的。相比之下，aiohttp中幾乎全部東西都是一個awaitable的協程，好比，session.request() 和 response.text(). 它是一個很棒的庫，可是在異步代碼中使用requests是有害的。

高層程序結構以下：

從本地文件url .txt中讀取url序列。
發送對URL的GET請求並解碼生成的內容。若是這失敗了，在那裏停下來找一個網址。
在響應的HTML中搜索href標記內的URL
將結果寫入foundurls.txt。
儘量異步和併發地執行上述全部操做。（對請求使用aiohttp，對文件附件使用aiofiles。這是IO的兩個主要示例，很是適合異步IO模型。）

下是urls.txt的內容。它並不龐大，而且主要包含高流量的網站：

$ cat urls.txt
https://regex101.com/
https://docs.python.org/3/this-url-will-404.html
https://www.nytimes.com/guides/
https://www.mediamatters.org/
https://1.1.1.1/
https://www.politico.com/tipsheets/morning-money
https://www.bloomberg.com/markets/economics
https://www.ietf.org/rfc/rfc2616.txt

列表中的第二個網址應該返回一個404響應，你須要優雅地處理這個響應。若是你正在運行此程序的擴展版本，你可能須要處理比這更多的問題，例如服務器斷開鏈接和無限重定向。

求自己應該使用單個會話進行，以充分利用會話的內部鏈接池。

讓咱們來看看完整的程序。以後，咱們將一步一步地介紹這些內容：

#!/usr/bin/env python3
# areq.py

"""Asynchronously get links embedded in multiple pages' HMTL."""

import asyncio
import logging
import re
import sys
from typing import IO
import urllib.error
import urllib.parse

import aiofiles
import aiohttp
from aiohttp import ClientSession

logging.basicConfig(
    format="%(asctime)s %(levelname)s:%(name)s: %(message)s",
    level=logging.DEBUG,
    datefmt="%H:%M:%S",
    stream=sys.stderr,
)
logger = logging.getLogger("areq")
logging.getLogger("chardet.charsetprober").disabled = True

HREF_RE = re.compile(r'href="(.*?)"')

async def fetch_html(url: str, session: ClientSession, **kwargs) -> str:
    """GET request wrapper to fetch page HTML.

    kwargs are passed to `session.request()`.
    """

    resp = await session.request(method="GET", url=url, **kwargs)
    resp.raise_for_status()
    logger.info("Got response [%s] for URL: %s", resp.status, url)
    html = await resp.text()
    return html

async def parse(url: str, session: ClientSession, **kwargs) -> set:
    """Find HREFs in the HTML of `url`."""
    found = set()
    try:
        html = await fetch_html(url=url, session=session, **kwargs)
    except (
        aiohttp.ClientError,
        aiohttp.http_exceptions.HttpProcessingError,
    ) as e:
        logger.error(
            "aiohttp exception for %s [%s]: %s",
            url,
            getattr(e, "status", None),
            getattr(e, "message", None),
        )
        return found
    except Exception as e:
        logger.exception(
            "Non-aiohttp exception occured:  %s", getattr(e, "__dict__", {})
        )
        return found
    else:
        for link in HREF_RE.findall(html):
            try:
                abslink = urllib.parse.urljoin(url, link)
            except (urllib.error.URLError, ValueError):
                logger.exception("Error parsing URL: %s", link)
                pass
            else:
                found.add(abslink)
        logger.info("Found %d links for %s", len(found), url)
        return found

async def write_one(file: IO, url: str, **kwargs) -> None:
    """Write the found HREFs from `url` to `file`."""
    res = await parse(url=url, **kwargs)
    if not res:
        return None
    async with aiofiles.open(file, "a") as f:
        for p in res:
            await f.write(f"{url}\t{p}\n")
        logger.info("Wrote results for source URL: %s", url)

async def bulk_crawl_and_write(file: IO, urls: set, **kwargs) -> None:
    """Crawl & write concurrently to `file` for multiple `urls`."""
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(
                write_one(file=file, url=url, session=session, **kwargs)
            )
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    import pathlib
    import sys

    assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
    here = pathlib.Path(__file__).parent

    with open(here.joinpath("urls.txt")) as infile:
        urls = set(map(str.strip, infile))

    outpath = here.joinpath("foundurls.txt")
    with open(outpath, "w") as outfile:
        outfile.write("source_url\tparsed_url\n")

    asyncio.run(bulk_crawl_and_write(file=outpath, urls=urls))

這個腳本比咱們最初的玩具程序要長，因此讓咱們把它分解一下。

常量HREF RE是一個正則表達式，用於提取咱們最終要搜索的HTML中的HREF標記：

>>> HREF_RE.search('Go to <a href="https://realpython.com/">Real Python</a>')
<re.Match object; span=(15, 45), match='href="https://realpython.com/"'>

協程 fetch html()是一個GET請求的包裝器，用於發出請求並解碼結果頁面html。它發出請求，等待響應，並在非200狀態的狀況下當即提出：

resp = await session.request(method="GET", url=url, **kwargs)
resp.raise_for_status()

若是狀態正常，則fetch_html（）返回頁面HTML（str）。值得注意的是，這個函數中沒有執行異常處理。邏輯是將該異常傳播給調用者並讓它在那裏處理：

html = await resp.text()

咱們等待session.request()和resp.text()，由於它們是awaitable的協程。不然，請求/響應週期將是應用程序的長尾、佔用時間的部分，可是對於異步輸入輸出，fetch_html()容許事件循環處理其餘可用的做業，例如解析和寫入已經獲取的URLs。

協程鏈中的下一個是parse()，它等待fetch html()獲取給定的URL，而後從該頁面的s html中提取全部的href標記，確保每一個標記都是有效的，並將其格式化爲絕對路徑。

誠然，parse()的第二部分是阻塞的，但它包括快速正則表達式匹配，並確保發現的連接成爲絕對路徑。

在這種特殊狀況下，這個同步代碼應該是快速和不明顯的。可是請記住，在給定的協程內的任何一行都會阻塞其餘協程，除非該行使用yield、await或return。若是解析是一個更密集的過程，您可能須要考慮使用executor()中的loop.run_in_executor()在本身的進程中運行這部分。

接下來，協程 write()接受一個文件對象和一個URL，並等待parse()返回一組已解析的URL，經過使用aiofiles(一個用於異步文件IO的包)將每一個URL及其源URL異步地寫入文件。

最後，bulk_crawl_and_write（）做爲腳本的協程鏈的主要入口點。它使用單個會話，併爲最終從urls.txt讀取的每一個URL建立任務。

這裏還有幾點值得一提:

默認的客戶機會話有一個最多有100個打開鏈接的適配器。要更改這一點，請將asyncio.connector.TCPConnector的實例傳遞給ClientSession。您也能夠按主機指定限制。
能夠爲整個會話和單個請求指定最大超時
此腳本還使用async with，它與異步上下文管理器一塊兒使用。我沒有專門討論這個概念，由於從同步到異步上下文管理器的轉換至關簡單。後者必須定義.__ aenter __（）和.__ aexit __（）而不是.__ exit __（）和.__enter__（）。正如您所料，async with只能在使用async def聲明的協程函數中使用。

若是您想進一步瞭解，GitHub上本教程附帶的文件有詳細的註釋。

下面是執行的所有榮耀，由於areq.py能夠在一秒鐘內獲取、解析和保存9個url的結果:

$ python3 areq.py
21:33:22 DEBUG:asyncio: Using selector: KqueueSelector
21:33:22 INFO:areq: Got response [200] for URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 115 links for https://www.mediamatters.org/
21:33:22 INFO:areq: Got response [200] for URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Got response [200] for URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.ietf.org/rfc/rfc2616.txt
21:33:22 ERROR:areq: aiohttp exception for https://docs.python.org/3/this-url-will-404.html [404]: Not Found
21:33:22 INFO:areq: Found 120 links for https://www.nytimes.com/guides/
21:33:22 INFO:areq: Found 143 links for https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Wrote results for source URL: https://www.mediamatters.org/
21:33:22 INFO:areq: Found 0 links for https://www.ietf.org/rfc/rfc2616.txt
21:33:22 INFO:areq: Got response [200] for URL: https://1.1.1.1/
21:33:22 INFO:areq: Wrote results for source URL: https://www.nytimes.com/guides/
21:33:22 INFO:areq: Wrote results for source URL: https://www.politico.com/tipsheets/morning-money
21:33:22 INFO:areq: Got response [200] for URL: https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Found 3 links for https://www.bloomberg.com/markets/economics
21:33:22 INFO:areq: Wrote results for source URL: https://www.bloomberg.com/markets/economics
21:33:23 INFO:areq: Found 36 links for https://1.1.1.1/
21:33:23 INFO:areq: Got response [200] for URL: https://regex101.com/
21:33:23 INFO:areq: Found 23 links for https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://regex101.com/
21:33:23 INFO:areq: Wrote results for source URL: https://1.1.1.1/

還不算太寒酸！做爲完整性檢查，你能夠檢查輸出的行數。在我作這個實驗的時候，它是626，但請記住，這可能會發生變更：

$ wc -l foundurls.txt
     626 foundurls.txt

$ head -n 3 foundurls.txt
source_url  parsed_url
https://www.bloomberg.com/markets/economics https://www.bloomberg.com/feedback
https://www.bloomberg.com/markets/economics https://www.bloomberg.com/notices/tos

下一步：若是你想增長難度，可讓這個網絡爬蟲進行遞歸。您可使用aio-redis跟蹤樹中已爬網的URL，以免請求它們兩次，並使用Python的networkx庫進行連接。記住要友好一點。將1000個併發請求發送到一個小的、毫無防備的網站是很是糟糕的。有一些方法能夠限制您在一個批處理中進行的併發請求數，例如使用asyncio的sempahore對象或使用相似這樣的模式。

上下文中的異步IO

既然您已經看到了至關多的代碼，讓咱們回過頭來考慮一下何時異步IO是一個理想的選擇，以及如何進行比較來得出這個結論，或者選擇其餘不一樣的併發模型。

什麼時候以及爲什麼異步IO是正確的選擇？

本教程不適用於異步IO與線程、多處理的擴展論述。然而，瞭解異步IO什麼時候多是三者中最好的候選是頗有用的。

關於異步IO與多處理之間的鬥爭實際上根本不是一場戰爭。事實上，它們能夠一塊兒使用。若是你有多個至關統一的CPU密集型任務（一個很好的例子是scikit-learn或keras等庫中的網格搜索），多進程應該是一個明顯的選擇。

若是全部函數都使用阻塞調用，那麼將async放在每一個函數以前不是一個好主意。(這實際上會下降你的代碼速度。)是正如前面提到的，異步IO和多處理能夠在一些地方和諧共存。

線程的伸縮性也比異步IO要差，由於線程是具備有限可用性的系統資源.在許多機器上建立數千個線程都會失敗，我不建議您首先嚐試它。建立數千個異步IO任務是徹底可行的。

當您有多個IO綁定任務時，異步IO會閃爍，不然任務將經過阻止IO密集等待時間來控制，例如：

網絡IO，不管您的程序是服務器端仍是客戶端
無服務器設計，例如點對點，多用戶網絡，如組聊天室
讀/寫操做，在這種操做中，您想要模仿「發射後無論」的風格，但沒必要擔憂鎖定正在讀寫的內容

不使用await的最大緣由是await只支持定義特定方法集的特定對象集。若是要對某個DBMS執行異步讀取操做，則不只須要查找該DBMS的Python包，這個包還必須支持python的async / await語法。包含同步調用的協程會阻止其餘協程和任務運行。關使用async / await的庫的列表，請參閱本教程末尾的列表。

Async IO It Is, but Which One?

本教程重點介紹異步IO，async / await語法，以及使用asyncio進行事件循環管理和指定任務。

asyncio固然不是惟一的異步IO庫。 Nathaniel J. Smith的觀察說了不少:

[在]幾年後，asyncio可能會發現本身淪落爲精明的開發人員避免使用的stdlib庫之一，好比urllib2。……實際上，我所說的是，asyncio是其自身成功的犧牲品：在設計時，它採用了可能的最好方法; 但從那之後，受asyncio啓發的工做 - 好比async / await的加入 - 已經改變了局面，讓咱們能夠作得更好，如今asyncio受到其早期承諾的束縛。via:(來源)【https://vorpus.org/blog/some-thoughts-on-asynchronous-api-design-in-a-post-asyncawait-world/】

儘管使用不一樣的api和方法，大名鼎鼎的curio 和 trio能作asyncio作的事情。就我的而言，我認爲若是你正在構建一箇中等規模，簡單的程序，只需使用asyncio就足夠了，並且易於理解，能夠避免在Python的標準庫以外添加另外一個大的依賴項。

但不管如何，看看curio和trio，你可能會發現他們用一種更直觀的方式完成了一樣的事情。此處介紹的許多與包不相關的概念也應該滲透到備用異步IO包中。

其餘零碎

在接下來的幾節中，您將看到asyncio和async/wait的一些雜項部分，這部分到目前爲止尚未徹底融入教程，可是對於構建和理解一個完整的程序仍然很重要。

其餘頂級asyncio 函數

除了asyncio.run（）以外，您還看到了一些其餘的包級函數，如asyncio.create_task（）和asyncio.gather（）。

您可使用create task()來調度協調程序對象的執行，後面跟着asyncio.run()

>>> import asyncio

>>> async def coro(seq) -> list:
...     """'IO' wait time is proportional to the max element."""
...     await asyncio.sleep(max(seq))
...     return list(reversed(seq))
...
>>> async def main():
...     # This is a bit redundant in the case of one task
...     # We could use `await coro([3, 2, 1])` on its own
...     t = asyncio.create_task(coro([3, 2, 1]))  # Python 3.7+
...     await t
...     print(f't: type {type(t)}')
...     print(f't done: {t.done()}')
...
>>> t = asyncio.run(main())
t: type <class '_asyncio.Task'>
t done: True

這種模式有一個微妙之處：若是你沒有在main（）中await t，它可能在main（）自己發出信號代表它已完成以前完成。由於在沒有await t 的狀況下asynio.run(main())調用loop.run_until_complete(main())，事件循環只關心main()是否完成了，而不是main()中建立的任務是否已經完成。沒有await t，循環的其餘事件可能在它們完成以前會被取消。若是須要獲取當前待處理任務的列表，可使用asyncio.Task.all_tasks（）。

注意：asyncio.create_task（）是在Python 3.7中引入的。在Python 3.6或更低版本中，使用asyncio.ensure_future（）代替create_task（）。

另外，還有asyncio.gather()。雖然它沒有作任何很是特殊的事情，可是gather()的目的是將一組協程(future)整齊地放到一個單一的future。所以，它返回一個單獨的future對象，若是await asyncio.gather（）並指定多個任務或協同程序，則表示您正在等待這些對象所有完成。(這與前面示例中的queue.join()有些類似。)gather()的結果將是跨輸入的結果列表：

>>> import time
>>> async def main():
...     t = asyncio.create_task(coro([3, 2, 1]))
...     t2 = asyncio.create_task(coro([10, 5, 0]))  # Python 3.7+
...     print('Start:', time.strftime('%X'))
...     a = await asyncio.gather(t, t2)
...     print('End:', time.strftime('%X'))  # Should be 10 seconds
...     print(f'Both tasks done: {all((t.done(), t2.done()))}')
...     return a
...
>>> a = asyncio.run(main())
Start: 16:20:11
End: 16:20:21
Both tasks done: True
>>> a
[[1, 2, 3], [0, 5, 10]]

你可能已經注意到gather（）等待您傳遞它的Futures或協程的整個結果集。或者，您能夠按完成順序循環遍歷asyncio.as_completed（）以完成任務。該函數返回一個迭代器，在完成任務時生成任務。下面coro([3,2,1])的結果將在coro([10,5,0])完成以前可用，而gather()的狀況並不是如此：

>>> async def main():
...     t = asyncio.create_task(coro([3, 2, 1]))
...     t2 = asyncio.create_task(coro([10, 5, 0]))
...     print('Start:', time.strftime('%X'))
...     for res in asyncio.as_completed((t, t2)):
...         compl = await res
...         print(f'res: {compl} completed at {time.strftime("%X")}')
...     print('End:', time.strftime('%X'))
...     print(f'Both tasks done: {all((t.done(), t2.done()))}')
...
>>> a = asyncio.run(main())
Start: 09:49:07
res: [1, 2, 3] completed at 09:49:10
res: [0, 5, 10] completed at 09:49:17
End: 09:49:17
Both tasks done: True

最後，你可能還能夠看到asyncio.ensure_future（）。你應該不多須要它，由於它是一個較低級別的管道API，而且很大程度上被後來引入的create_task（）取代。

await的優先級

雖然它們的行爲有些類似，但await關鍵字的優先級明顯高於yield。這意味着，因爲它的綁定更緊密，在不少狀況下，您須要在yield from語句中使用括號，而在相似的await語句中則不須要。有關更多信息，請參見PEP 492中的await表達式示例。

總結

你如今已經準備好使用async / await和它構建的庫了。如下是你已經學到的的內容概述：

異步IO做爲一種與語言無關的模型，經過讓協程彼此間進行間接通訊來實現併發
Python中用於標記和定義協程的新關鍵字async、await的一些細節。
提供用於運行和管理協程的API的Python包asyncio

附加資源

Python版本細節

Python中的異步IO發展迅速，很難跟蹤何時發生了什麼。下面列出了與asyncio相關的Python小版本更改和介紹：

3.3： yield from表達式容許生成器委派
3.4：asyncio以臨時API狀態引入Python標準庫
3.5：async和await成爲Python語法的一部分，用於表示和等待協程。它們尚未成爲保留關鍵字（您仍然能夠定義名爲async和await的函數或變量）。
3.6：引入異步生成器和異步理解/鏈、推導。asyncio的API被聲明爲穩定的，而不是臨時的。
3.7：async和await成爲保留關鍵字(它們不能用做標識符。)。它們用於替換asyncio.coroutine()裝飾器。asyncio.run（）被引入asyncio包，其中包括許多其餘功能。

若是您想要安全（而且可以使用asyncio.run（）），請使用Python 3.7或更高版原本獲取完整的功能集。

如下是其餘資源的精選列表：

Real Python: 使用併發加速Python程序
Real Python:什麼是Python全局解釋器鎖?
CPython: asyncio包源碼
Python docs: 數據模型> 協程
TalkPython:Python中的異步技術和示例
Brett Cannon: 在Python 3.5中Async-Await究竟是如何工做的
PYMOTW: asyncio
A. Jesse Jiryu Davis and Guido van Rossum: 使用asyncio協程的Web爬蟲
Andy Pearce: Python協程的狀態:yield from
Nathaniel J. Smith: post-async/await世界中異步API設計的幾點思考
Armin Ronacher: 我不理解Python的Asyncio
Andy Balaam: asyncio系列 (4 篇)
Stack Overflow: async-await函數中的Python asyncio.semaphore
Yeray Diaz:
- 面向工做的Python開發人員的AsyncIO
- Asyncio協程模式：超越await

Python文檔的 What’s New 部分更詳細地解釋了語言變化背後的動機：

Python 3.3新變更 (yield from 和PEP 380)
Python 3.6新變更 (PEP 525 & 530)

來自David Beazley的：

YouTube 視頻:

PEP	建立時間
PEP 342 – 經過加強型生成器的協程	2005-05
PEP 380 – 委託給子生成器的語法	2009-02
PEP 3153 – 異步IO支持	2011-05
PEP 3156 – 異步IO支持從新啓動：「asyncio」模塊	2012-12
PEP 492 – async和await語法的協程	2015-04
PEP 525 – 異步生成器	2016-07
PEP 530 – Asynchronous Comprehensions	2016-09

使用async/await的庫

來自 aio-libs:

aiohttp: 異步HTTP客戶端/服務器框架
aioredis: 異步IO Redis支持
aiopg: 異步IO PostgreSQL 支持
aiomcache: 異步IO memcached 客戶端
aiokafka: 異步IO Kafka 客戶端
aiozmq: 異步IO ZeroMQ 支持
aiojobs:用於管理後臺任務的做業調度程序
async_lru: 用於異步IO的簡單LRU緩存

來自 magicstack:

uvloop:超快的異步IO事件循環
asyncpg: (也很是快)異步IO PostgreSQL支持

來自其餘:

trio: 更友好的「asyncio」，旨在展現一個更加簡單的設計
aiofiles: 異步文件 IO
asks: 異步類requests的http 庫
asyncio-redis: 異步IO Redis 支持
aioprocessing: 將multiprocessing模塊與asyncio集成在一塊兒
umongo: 異步IO MongoDB 客戶端
unsync: Unsynchronize asyncio
aiostream:相似'itertools'，但異步

個人感想：

其實讀完這篇文章，我相信有不少人仍舊會有困惑——異步IO底層究竟是怎麼實現的？早些時候我也很困惑，要說多線程多進程咱們很好理解，由於咱們知道經常使用的現代計算機是根據時間片分時運行程序的。到了異步IO或者協程這裏居然會出現一段沒有CPU參與的時間。在我學習Javascript/nodejs的時候就更困惑了，web-base的javascript和backend nodejs都是單線程設計的，它的定時器操做怎麼實現的？它的界面異步操做怎麼實現的？後來讀了《UNIX環境高級編程》纔有種「恍然大悟」的感受。在學習編程語言的時候，每每認爲語言自己是圖靈完備的，編程語言設定的規則就是整個世界。但實際上，編程語言的圖靈完備僅體如今邏輯和運算上，其餘的一些設施底層不是語言自己就可以徹底解釋的。咱們，至少是我本身，在學習一個語言工具的時候每每忽略了一個早就知道的現實——現代常規的編程，都是面向操做系統的編程！不管是多線程、多進程仍是異步IO自己都是操做系統提供的功能。多餘web-base的javascript更是面向瀏覽器編程。瀏覽器不提供異步IO相關的功能，Web-base 的javascript自己是沒辦法實現的，操做系統不支持異步IO，什麼語言也不行～golang的go程也不過是從系統手中接管了生成線程以後的再分配管理。正像Linux/Unix編程標準是兩個的合體——ANSI C + POSIX，咱們學習的語言正對應ANSI C，但多線程、多進程、信號這些東西自己不是語言規範裏面的，他們是POSIX裏的，是操做系統的規範，是操做系統提供的！再進一步，爲何操做系統能實現？由於硬件支持這樣的實現！