在Python中使用Asyncio系統（3-1）Asyncio之塔

演練Async

Python中的asyncio API是複雜的，因為它旨在為不同的群體解決不同的問題。不幸的是，幾乎沒有什麼指導意見可以幫助你弄清楚asyncio的哪些部分對你所在的群體是重要的。

我的目標就是幫你解決這個問題。Python中的非同步特性有兩個主要的目標受眾：

終端使用者開發人員一些希望使用asyncio建立應用程式的人。我假設你在這個群體裡。

框架開發人員一些希望製作終端使用者開發人員可以在其應用程式中使用的框架和庫的人。

現在社群中關於asyncio的許多困惑是由於缺乏對這一差異的理解。例如，針對asyncio的官方Python文件更適合框架開發人員而不是終端使用者。這意味著終端使用者開發人員在閱讀這些文件時很快就會被這些明顯的複雜性所震驚。在你能夠使用Asyncio之前，你有點被迫接受這個複雜度。

我希望這本書可以幫助您區分Asyncio中對終端使用者開發者重要的特性和對框架開發者重要的特性。

我的目標是讓你只對asyncio的構件有最基本的理解——足以讓你能夠用它編寫簡單的程式，當然也足以讓你能夠深入研究更完整的參考資料。

（在編寫本書時，Asyncio的參考資料只有官方Python文件中的API規範和一系列的部落格文章，其中有幾篇在本書連結中。）

首先，我們有一個“快速入門”部分，介紹了Asyncio應用程式最重要的構件。

快速入門

鑽研Asyncio的官方文件是相當可怕的。有很多章節都有新的、神秘的詞彙和概念，即使是有經驗的Python程式設計師也會感到陌生，因為Asyncio在Python中是一個非常新的東西。稍後我將詳細介紹這些內容，並解釋如何處理asyncio模組的文件，但現在你需要知道，使用asyncio庫需要擔心的實際困難比看起來要小得多。

Yury Selivanov， PEP 492的作者和非同步Python的主要貢獻者，在他的PyCon 2016演講“Python 3。5中的async/await以及它為什麼很棒”中解釋的，asyncio模組中的許多api實際上是為框架設計者，而不是終端使用者開發人員準備的。在那次演講中，他強調了終端使用者應該關注的主要功能。這些是整個asyncio API的一個小子集，可以總結如下：

啟動asyncio的事件迴圈

呼叫async/await函式

建立一個在事件迴圈上執行的任務

等待多個任務完成

在所有併發任務完成後關閉迴圈

在這一節中，我們要研究這些核心特性，還要看看怎麼在Python中立即開始基於事件的程式設計。

Python中的Asyncio的“Hello World”看起來像示例3-1

示例3-1 Asyncio的Hello World

# quickstart。pyimport asyncio， timeasync def main（）： print（f‘{time。ctime（）} Hello！’） await asyncio。sleep（1。0） print（f‘{time。ctime（）} Goodbye！’）asyncio。run（main（））

（L9） asyncio提供了一個run（）函式來執行async def函式和從那裡呼叫的所有其他協程，比如main（）函式中的sleep（）。

這是示例3-1的執行輸出

$ python quickstart。pySun Aug 18 02：14：34 2019 Hello！Sun Aug 18 02：14：35 2019 Goodbye！

在實踐中，大多數基於asyncio的程式碼會使用這裡顯示的run（）函式，但重要的是瞭解更多關於該函式為你做了什麼。這種理解很重要，因為它將影響你如何設計更大的應用程式。

例3-2是我稱之為“Hello-ish World”的例子。它與run（）所做的並不完全相同，但已經足夠引入我們將在本書的其餘部分中構建的思想。你將需要協同例程的基本知識（將在本章後面進行深入討論），但是現在請嘗試繼續學習並集中於高階概念。

示例3-2 Asyncio的Hello-ish World

# quickstart。pyimport asyncioimport timeasync def main（）： print（f“{time。ctime（）} Hello！”） await asyncio。sleep（1。0） print（f“{time。ctime（）} Goodbye！”）loop = asyncio。get_event_loop（） task = loop。create_task（main（）） loop。run_until_complete（task） pending = asyncio。all_tasks（loop=loop）for task in pending： task。cancel（）group = asyncio。gather（*pending， return_exceptions=True） loop。run_until_complete（group） loop。close（）

loop = asyncio。get_event_loop（）在執行任何協程之前，你需要一個事件迴圈例項，這段程式碼就是給你展示如何建立一個事件迴圈。事實上，無論你在哪裡呼叫它，get_event_loop（）每次都會給你相同的迴圈例項，只要你只使用一個執行緒。（asyncio API允許使用多個迴圈例項和執行緒做許多瘋狂的事情，但這不是正確的做法。99

%

的情況下，你的應用只會使用一個主執行緒，就像現在程式碼所示。）如果你在一個async def函式中，你應該呼叫asyncio。get_running_loop（），它總是給你你所期望的。這在本書後面有更詳細的介紹。

task = loop。create_task（coro）在本例中，具體呼叫的是loop。create_task（main（））。在你這樣做之前，協程函式不會被執行。create_task（）排程你的協程在迴圈中執行。（使用引數名coro是API文件中的一種常見約定。它指的是協程；也就是說，嚴格來說，是呼叫async def函式的結果，而不是函式本身。）返回的任務物件可用於監視任務的狀態（例如，它是仍在執行還是已經完成），還可用於從已完成的協程中獲取結果值。可以使用task。cancel（）來取消任務。

loop。run_until_complete（coro）這個呼叫將阻塞當前執行緒，而當前執行緒通常是主執行緒。請注意，run_until_complete（）將保持主迴圈執行只到給定的coro完成——當然所有排程在主迴圈上的其他任務也將在主迴圈執行時保持執行。在內部，asyncio。run（）會呼叫run_until_complete（），從而以相同的方式阻塞主執行緒。

group = asyncio。gather（task1， task2， task3）當程式的“主”部分解除阻塞時，無論是由於接收到程序訊號，還是由於呼叫loop。stop（）的程式碼停止了迴圈，run_until_complete（）之後的程式碼都將保持執行。這裡顯示的標準習慣用法是收集仍然掛起的任務，取消它們，然後再次使用loop。run_until_complete（），直到這些任務完成。Gather（）是用於進行收集的方法。請注意，asyncio。run（）將執行所有的取消、收集和等待掛起的任務完成。

loop。close（）loop。close（）通常是最後一個動作：在停止迴圈時必須呼叫它，它將清除所有佇列並關閉執行器。停止的迴圈可以重新啟動，但關閉的迴圈將永遠消失。在內部，asyncio。run（）將在返回之前關閉迴圈。這很好，因為每次呼叫run（）都會建立一個新的事件迴圈。

示例3-1表明，如果使用asyncio。run（），這些步驟都是不必要的：它們都已幫你完成。然而，理解這些步驟很重要，因為在實踐中會出現更復雜的情況，你需要額外的知識來處理它們。其中一些將在本書後面詳細介紹。

Python中的asyncio暴露了圍繞事件迴圈的大量底層機制，並要求您注意生命週期管理等方面。這與Node。js不同。例如，Node。js也包含一個事件迴圈，但被隱藏起來。而且，一旦使用了一點asyncio，就會開始注意到啟動和關閉事件迴圈的模式與本文給出的程式碼相差不大。我們將在本書後面更詳細地研究管理迴圈生命週期的一些細微差別。

我在前面的例子中遺漏了一些東西。你需要了解的最後一項基本功能是如何執行阻塞函式。關於協作多工的事情是，你需要所有I/O繫結函式…好吧，協作，這意味著允許上下文切換回使用關鍵字await的迴圈。目前市面上的大多數Python程式碼都不會這樣做，而是依賴於您線上程中執行這些函式。在非同步def函式得到更廣泛的支援之前，您會發現使用這樣的阻塞庫是不可避免的。

為此，asyncio提供了一個與併發中的API非常相似的API。futures包。這個包提供了一個ThreadPoolExecutor和一個ProcessPoolExecutor。預設是基於執行緒的，但是可以使用基於執行緒或基於執行緒池的executor。我省略了前面例子中的執行器考慮，因為它們會模糊基本部件如何組合在一起的描述。現在這些都講完了，我們可以直接看executor了。

有幾個技巧需要注意。讓我們看一下示例3-3中的程式碼示例。

示例3-3 基本的executor

# quickstart_exe。pyimport timeimport asyncioasync def main（）： print（f‘{time。ctime（）} Hello！’） await asyncio。sleep（1。0） print（f‘{time。ctime（）} Goodbye！’）def blocking（）： time。sleep（0。5） print（f“{time。ctime（）} Hello from a thread！”）loop = asyncio。get_event_loop（）task = loop。create_task（main（））loop。run_in_executor（None， blocking） loop。run_until_complete（task）pending = asyncio。all_tasks（loop=loop） for task in pending： task。cancel（）group = asyncio。gather（*pending， return_exceptions=True）loop。run_until_complete（group）loop。close（）

（L10）Blocking（）在內部呼叫傳統的time。sleep（），這會阻塞主執行緒並阻止您的事件迴圈執行。這意味著您不能將此函式作為協程——實際上，您甚至不能從主執行緒的任何地方呼叫此函式，而主執行緒正是asyncio迴圈執行的地方。我們透過在執行器中執行這個函式來解決這個問題。

（L11） 與此部分無關，但是在本書後面要記住一些東西：注意main（）協程中的阻塞sleep時間（0。5秒）比非阻塞sleep時間（1秒）要短。這使得程式碼示例非常整潔。在第68頁的“在關機期間等待Executor”中，我們將探討如果Executor函式在關機期間比非同步函式活得長會發生什麼。

（L17）await loop。run_in_executor（None， func）這是asyncio的最後一個基本的、必須知道的特性。有時候你需要在一個單獨的執行緒甚至是一個單獨的程序中執行一些東西；這個方法就是用來做這個的。這個例子裡我們要把阻塞函式傳遞給預設的執行器。

（不幸的是，run_in_executor（）的第一個引數是要使用的Executor例項，為了使用預設值，必須傳遞None。每次我使用這個引數時，都感覺“executor”引數是一個預設值為None的關鍵詞引數。）注意run_in_executor（）不會阻塞主執行緒：它只調度執行器任務的執行（它返回一個Future，這意味著如果該方法在另一個協程函式中被呼叫，就可以await它）。只有在呼叫run_until_complete（）之後，執行器任務才會開始執行，事件迴圈才開始處理事件。

進一步說明標註2：pending中的任務集不包括在run_in_executor（）中對blocking（）的呼叫條目。對於任何返回Future而不是Task的呼叫都是如此。文件很擅長指定返回型別，你會看到返回型別；只要記住all_tasks（）實際上只返回Tasks，而不是Futures。

下面是執行這個指令碼的輸出：

$ python quickstart_exe。py Sun Aug 18 01：20：42 2019 Hello！ Sun Aug 18 01：20：43 2019 Hello from a thread！ Sun Aug 18 01：20：43 2019 Goodbye！

現在您已經瞭解了asyncio滿足終端使用者開發人員需求的最基本部分，現在是時候擴充套件我們的範圍並將asyncio API安排成一種層級結構了。這會讓你更容易理解和消化如何從文件中獲取您所需要的內容，這一節的內容就這些了。

Asyncio之塔

正如您在前一節中看到的，作為終端使用者開發人員使用asyncio只需要知道少量命令。不幸的是，asyncio的文件提供了大量的api，而且它以一種非常“扁平”的格式進行，這使得很難區分哪些是通用的，哪些是提供給框架設計人員的工具。

當框架設計人員檢視相同的文件時，他們會尋找可以連線新框架或第三方庫的關注點。在本節中，我們將從框架設計人員的角度來研究asyncio，以瞭解它們如何構建新的非同步相容庫。希望這將有助於進一步劃分出你在自己的工作中需要關注的特性。

從這個角度來看，將asyncio模組看成一個層級結構（而不是一個平面列表）可能更容易理解，在這個層級結構中，每一層都構建在前面一層的規範之上。不幸的是，它並沒有那麼整潔，我對錶3-1中的排列做了一些改動，希望這能給你提供關於asyncio API的另一種視角。

表3-1 Asyncio的特性按層級排列；對於終端使用者的開發者來說，更需要關注表中的粗體字部分。

在最基本的層次，第1層，是你在本書前面已經看到的協程。這是可以開始考慮設計第三方框架的最低級別，令人驚訝的是，這在當前可用的兩個（不是一個，而是兩個）非同步框架中相當流行：Curio和Trio（與Asyncio類似的非同步庫）。這兩個框架都只依賴於Python中的本地協程，而不依賴於asyncio庫。

下一層是事件迴圈。協程本身是沒有用的：如果沒有一個迴圈來執行它們，它們就不會做任何事情（因此，Curio和Trio必須實現它們自己的事件迴圈）。asyncio提供了一個迴圈規範AbstractEventLoop和一個實現BaseEventLoop。

規範和實現之間的明確分層使得第三方開發人員可以對事件迴圈進行替代實現，uvloop就是這樣做的，它提供了比asyncio標準庫中更快的迴圈實現。重要的是，uvloop只是簡單地“插入”層級結構，並只替換了層級的迴圈部分。做出這些選擇，正是asyncio API被設計成這樣的原因，各活動部分之間有明確的分層。

第3層和第4層是非常密切相關的future和task；它們被分開只是因為Task是Future的子類，但它們經常被認為是在同一層。Future例項表示某種正在進行的操作，該操作將透過事件迴圈的通知返回結果，而Task表示在事件迴圈上執行的協程。簡而言之：future是“迴圈感知的”，而task則是“迴圈感知的”和“協程感知的”。作為一名終端使用者開發人員，你要處理的task遠遠多於future，但對於框架設計人員來說，比例則可能是相反的，這取決於細節。

第5層表示啟動和等待必須在單獨執行緒甚至是單獨程序中執行的工作的設施。

第6層表示其他非同步感知工具，如asyncio。Queue。我本可以把這一層放在網路層之後，但我認為，在我們討論I/O層之前，先解決所有協程感知的api更整潔。asyncio提供的Queue具有與Queue模組中的執行緒安全Queue非常相似的API，除了asyncio版本需要在get（）和put（）上使用await關鍵字。你不能在協程中直接使用queue。Queue，因為它的get（）會阻塞主執行緒。

最後是網路I/O層，7到9。作為一個終端使用者開發人員，最方便使用的API是第9層的streams API。我把streams API定位在塔的最高抽象級別。緊接在它下面（第8層）的協議層API，是一個更細粒度的API；你可以在所有可能使用streams層的情況下使用協議層，但使用streams會更簡單。最後一個網路I/O層是傳輸層（第7層）。除非您正在建立一個框架供其他人使用，並且需要自定義如何設定傳輸層，否則不太可能直接使用這個層。

在第22頁的“快速入門”中，我們看了一個人開始使用asyncio庫所需的絕對最低限度的知識。現在我們已經瞭解了整個asyncio庫API是如何組合在一起的，我想重新審視這個特性列表，並再次強調可能需要學習的部分。

這些是在學習如何使用asyncio庫模組編寫網路應用時最需要關注的層級：

第1層

理解如何編寫async def函式和使用await呼叫和執行其他協程是至關重要的。

第2層

瞭解如何啟動、關閉和與事件迴圈互動是至關重要的。

第5層

executor是在非同步應用程式中使用阻塞程式碼所必需的（譯者：把阻塞部分放到一個單獨的執行器中），因為大多數第三方庫還不是非同步相容的。一個常見例子是資料庫ORM庫SQLAlchemy，對於asyncio，目前還沒有類似的庫可供選擇（譯者：現在SQLAlchemy好像已經支援非同步了）。

第6層

如果您需要將資料提供給一個或多個長時間執行的協程，最好的方法是使用asyncio。Queue。用於線上程之間分發資料的場景，這與使用queue。Queue的策略完全相同。Asyncio版本的Queue使用與標準庫佇列模組相同的API，只是使用協程會替換掉一些類似get（）這樣的阻塞方法。

第9層

streams API提供了透過網路處理socket通訊的最簡單方法，你應該從這裡開始構建網路應用程式的原型。您可能會發現需要更細粒度的控制，然後您可以切換到協議API，但在大多數專案中，通常最好保持事情簡單，直到確切地知道要解決的問題是什麼。

當然，如果你正在使用一個asyncio相容的第三方庫來為你處理所有的socket通訊，比如aiohttp，你根本就不需要直接使用asyncio網路層。在這種情況下，必須高度依賴庫提供的文件。

asyncio庫試圖為終端使用者開發人員和框架設計人員都提供足夠的特性。不幸的是，這意味著asyncio API可能會顯得有點雜亂無章。我希望這一節已經提供了足夠的路線圖來幫助你挑選你所需要的部件。

在下一節中，我們將更詳細地檢視前面層級列表中的元件部分。

========================

以上是第3章的前兩節，第3章比較大，我只能分為幾個節一起出一期了。不然一次的閱讀量太大了。