了解 Python 中的異步程式設計 了解 Python 中的異步程式設計

<b>本文講的是了解 Python 中的異步程式設計，</b>

<b></b>

如何使用 Python 來編寫異步程式以及為什麼你需要做這件事。

我們中的大部分人一開始寫的都是同步程式，這種類型的程式可以被認為是在同一時間隻運作一個執行步驟，一步接着一步這樣相繼執行的。

即使是有條件分支、循環和函數調用，我們仍然可以認為代碼在同一時間隻執行一步，當這一步完成時，才執行下一步。

下面是使用這種模式運作的示例程式：

批量處理程式通常是寫成同步程式的：擷取一些輸入，處理它，然後建立一些輸出。按照邏輯一步接一步地運作直到得到我們想要的輸出。除了這些執行的步驟和順序外，這類程式并不需要關注其它任何事情。

指令行程式通常是一個小而快的程式，用來把一些東西“轉換“成其他一些東西。這個程式表現出來就是一系列程式步驟連續執行直到完成任務。

一個異步程式表現的就不一樣。它仍然是每次隻執行一步。然而其不同之處在于系統可能不會等到一個執行步驟完成後再執行下一步。

這意味着即使先前的執行步驟（或者多個步驟）還在“其他地方“執行，我們仍然可以繼續執行程式的下一步。這也意味着當在"其他地方"的步驟執行完成時，我們程式代碼中必須去處理它。

為什麼我們想要以這種方式寫程式呢？簡單來說就是它可以幫助我們處理特定類型的程式設計問題。

這裡有個概念性的程式或許可以作為認識異步程式設計的例子：

它的基本工作單元和我們之前描述的批量處理程式相同；擷取一些輸入，處理這些輸入，然後建立輸出。通過寫一個同步程式就可以建立一個工作的 web 伺服器。

它将是一個絕對糟糕的 web 伺服器。

為什麼？ 在一個 web 伺服器的情況下，一個工作單元（輸入、處理、輸出）不是這個伺服器唯一的目的。它真正的目的是能長時間同時處理數百上千個工作單元。

我們能讓這個同步伺服器變得更好嗎？ 當然，我們可以通過優化我們的執行步驟來讓它們運作的盡可能快。但是不幸的是，這種方法有非常現實的限制，進而導緻 web 伺服器無法足夠快的響應請求，并且也不能處理足夠數量的目前使用者。

上述方法優化的真實限制是什麼？ 網絡的速度，檔案的讀寫速度，資料庫的查詢速度，其它連接配接服務的速度等。這個清單的的共有特征是它們都是 IO 函數。所有這些項的處理速度都比我們的 CPU 處理速度要慢很多個數量級。

在一個同步程式裡如果一個執行步驟開始一次資料庫查詢（比如說），在這次查詢傳回一些資料之前，CPU 本質上會空閑很長一段時間，之後它才可以繼續運作下一個執行步驟。

對于面向批量處理的程式這并不是關鍵，它的目的是處理通過 IO 操作得到的結果，而且這個過程所花時間通常比 IO 操作長得多。任何優化的工作都将側重于處理的工作而不是 IO。

檔案，網絡和資料庫 IO 操作都很快，但是它們的執行速度仍然比 CPU 執行速度慢。異步程式設計技術讓我們的程式可以利用相對較慢的 IO 處理來釋放 CPU 進而執行其他工作。

當我開始嘗試了解異步程式設計時，我咨詢的人們和查閱的文檔談了很多編寫非阻塞代碼的重要性。是的，這些從來沒有幫助到我。

什麼是非阻塞代碼？什麼是阻塞代碼？這個資訊就像我們有一個參考手冊，但是手冊裡面沒有具有實際意義的内容，來描述如何有意義地使用這些技術細節。

相較于同步程式，編寫異步程式是不一樣的，讓你了解起來會有一點困難。這就很有趣了，因為無論是在我們生活的世界裡，還是我們與之互動的方式，這些幾乎都是完全異步的。

這裡有一個大多數人都相關聯的例子： 作為一個父母嘗試同時做好幾件事；包括平衡支票本、洗衣服和照看孩子。

我們做這些事的時候甚至從來都沒有細想，但是現在讓我們試着把它們拆分出來：

平衡支票本是一個我們正在嘗試将其完成的任務，而且我們可以把它看作一個同步任務；一步接着另一步執行直至任務完成。

但是，我們可以離開這個任務去洗衣服，把烘幹機裡已經烘幹的衣物取出，再把已經洗完的衣服從洗衣機拿出來之後放到烘幹機裡并開始把另一些未洗的衣服放入洗衣機。不管怎樣，這些任務是可以被異步完成的。

雖然我們在使用洗衣機和烘幹機來洗衣服,這個過程是一個同步任務而且我們正在處理該任務，但是洗衣服的大部分任務是發生在我們啟動洗衣機和烘幹機之後發生的，這時候我們已經離開并傳回平衡支票本的任務。現在任務就是異步的了，洗衣機和烘幹機将會獨立運作，一直到其中任意一個需要我們去處理時蜂鳴器就會響。

看孩子是另一個異步的任務。一旦他們起床了并且在玩耍，他們是一個人在那玩（一定程度上）直到他們需要我們的注意；一些孩子餓了，一些受傷了，一些在大聲的叫喊，作為父母我們需要對此作出響應。照看孩子是一個長期運作的高優先級任務，重要性超過了任何我們可能正在做的其他任務，比如平衡支票本和洗衣服。

這個例子展示了阻塞和非阻塞代碼。比如說當我們去洗衣服的時候，CPU（父母）就是忙碌的并且阻塞執行其它的工作。

但是沒有關系，因為 CPU 正在忙碌而且這個任務運作時間相對來說是比較快的，當我們啟動洗衣機和烘幹機之後傳回做其他事時，這個洗衣的任務現在就變成異步的了，因為 CPU 正在做其它的事情，如果你願意，這時候已經改變了運作的上下文，而且當洗衣任務完成時你将通過機器的蜂鳴器得到通知。

作為人類這是我們工作的方式，我們很自然的在同一時間做多件事情，這過程經常是不加思索的。作為程式員，其中的訣竅就是把這種行為轉化為做同樣事的代碼。

讓我們嘗試使用你可能熟悉的代碼觀念來"程式設計"：

想想嘗試使用完全同步的方式來完成這些任務。在這種情形下，如果我們是好的父母，我們就會一直照看着孩子，等待孩子這邊有一些需要我們關注的事情發生。在這種情況下我們不會做其他任何事，比如平衡支票本或者洗衣服。

我們可以按任何我們想要的方式在确定任務的優先級，但是在同一時間隻有一個任務以同步的方式發生，以一個接着另一個的方式。這種方式就像先前描述的同步伺服器一樣，它的确可以工作，但這将是一種可怕的運作方式。

直到孩子睡着之前，我們都不能幹其它任何事，其他所有事隻能在這個之後才能做了，但是這時候已經夜幕降臨了。這樣的一個星期之後，大多數父母會選擇跳出窗外。

讓我們改變上面的方式，使用輪詢來完成多件事。在這個方式中，父母會周期性的離開任何目前正在做的任務，去檢查是否有其它任務需要注意。

由于我們正在對一個父母程式設計，是以讓我們來設定這個輪詢的間隔時間，比如說15分鐘。是以之後每隔15分鐘，這個父母就會去檢查洗衣機，烘幹機或者孩子是否需要注意，然後再傳回去處理平衡支票本。如果其中的任何一件事需要注意，父母就需要先完成這個工作再傳回處理平衡支票本，之後繼續進行輪詢的循環。

如果這樣做，任務就都可以完成，但是這樣仍會有一些問題。CPU（父母）花了大量時間來檢查不需要注意的事，僅僅是因為這些事還沒有被完成，比如說像洗衣機和烘幹機。給定輪詢的時間間隔，在這段時間内任務執行完成是完全有可能的，但是在輪詢時間 15 分鐘到達之前，該任務有一段時間是不會得到注意的。對于照看孩子這個高優先級的任務，當一些非常嚴重的事情發生時，這個 15 分鐘可能的視窗期是讓人不能忍受的。

我們可以通過縮短我們的輪詢時間來解決這個問題，但是現在 CPU 甚至會花費更多的時間在任務之間進行上下文切換，并且我們得到的收益開始逐漸降低。當我們像這樣生活了幾個星期之後，下場可以參考我之前關于視窗和跳躍的評論。

我們經常可以從作為父母的人口中聽到"隻有把自己克隆了才能完成這麼多的事"。由于現在我們假裝可以對父母這個角色程式設計，我們可以通過使用線程來實作克隆。

如果我們把所有的任務看成一個"程式"，我們就可以把這些任務分解出來并且使用線程來運作這些任務，隻要克隆這個父母就可以了。現在對于每個任務都有一個父母執行個體；包括照看孩子，看管洗衣機，看管烘幹機和平衡收支本，所有這些任務都是獨立運作的。這對于這個程式的問題來說聽起來是一個很棒的解決方案。

但是确實是這樣嗎？因為我們必須明确的告訴這些父母執行個體（CPUs）在程式裡面要做什麼，當所有的執行個體都共享程式空間内的全部資源時，我們就會遇到一些問題。

比如說，當監控烘幹機的父母看到衣服已經烘幹了，就會去控制烘幹機并開始把裡面的衣服取出來。當負責烘幹機的父母正在取出衣服時，負責洗衣機的父母看到洗衣機也洗完衣服了，就會去控制洗衣機，取出衣服後就會想要去控制烘幹機來将洗完的衣服從洗衣機放到烘幹機。這時候控制烘幹機的父母已經從烘幹機取出衣服而想要去控制洗衣機，并且之後會把衣服從洗衣機移動到烘幹機。

他們兩個都控制着自己的資源并且希望控制對方的資源。他們就會一直等待對方釋放對資源的控制權。作為程式員，我們必須編寫代碼來處理這種情況。

這裡是另一個因為父母線程可能引發的問題。比如不幸的是，一個孩子受傷了，父母就必須要帶孩子去緊急治療。因為現在這個父母的克隆是專門用于照看孩子的，是以可以馬上響應。但是在這個緊急情況下，照看孩子的父母必須寫一張相當大的支票來支付緊急護理的自付額。

同時，在支票本上工作的父母并不知道已經寫了這個大額的支票，這時候家庭的賬戶就已經透支了。因為所有父母的克隆都在同一個程式内工作，家庭的錢（支票本）是這個程式世界裡的一個共享資源，我們需要想出一個辦法讓照看孩子的父母可以通知平衡支票本的父母發生了什麼。或者就要提供某種鎖機制，這樣在同一時間隻有一個父母執行個體能更新這個資源。

所有這些事情都是可以在程式的線程代碼中管理的，但是很難把代碼寫正确并且當出現問題時也很難 debug。

現在我們将采用在"頭腦風暴"中概述的一些方法，我們将把它們轉變為可以運作的 Python 代碼。

第一個例子展示的是一種有些刻意設計的方式，即有一個任務先從隊列中拉取"工作"之後再執行這個工作。在這種情況下，這個工作的内容隻是擷取一個數字，然後任務會把這個數字疊加起來。在每個計數步驟中，它還列印了字元串表明該任務正在運作，并且在循環的最後還列印出了總的計數。我們設計的部分即這個程式為多任務處理在隊列中的工作提供了很自然的基礎。

該程式中的"任務"就是一個函數，該函數可以接收一個字元串和一個隊列作為參數。在執行時，它會去看隊列裡是否有任何需要處理的工作，如果有，它就會把值從隊列中取出來，開啟一個 for 循環來疊加這個計數值并且在最後列印出總數。它會一直這樣運作直到隊列裡什麼都沒剩了才會結束離開。

當我們在執行這個任務時，我們會得到一個清單表明任務一（即代碼中的 task One）做了所有的工作。它内部的循環消費了隊列裡的全部工作，并且執行這些工作。當退出任務一的循環後，任務二（即代碼中的 task Two）有機會運作，但是它會發現隊列是空的，因為這個影響，該任務會列印一段語句之後退出。代碼中并沒有任何地方可以讓任務一和任務二協作的很好并且可以在它們之間切換。

程式（<code>example_2.py</code>）的下個版本通過使用生成器增加了兩個任務可以跟好互相協作的能力。在任務函數中添加 yield 語句意味着循環會在執行到這個語句時退出，但是仍然保留當時的上下文，這樣之後就可以恢複先前的循環。在程式後面 "run the tasks" 的循壞中當 <code>t.next()</code> 被調用時就可以利用這個。這條語句會在之前生成（即調用 yield 的語句處）的地方重新開始之前的任務。

這是一種協作并發的方式。這個程式會讓出對它目前上下文的控制，這樣其它的任務就可以運作。在這種情況下，它允許我們主要的 "run the tasks" 排程器可以運作任務函數的兩個執行個體，每一個執行個體都從相同的隊列中消費工作。這種做法雖然聰明一些，但是為了和第一個示例達成同樣結果的同時做了更多的工作。

當程式運作時，輸出表明任務一和任務二都在運作，它們都從隊列裡消耗工作并且處理它。這就是我們想要的，兩個任務都在處理工作，而且都是以處理從隊列中的兩個項目結束。但是再一次，需要做一點工作來實作這個結果。

這裡的技巧在于使用 <code>yield</code> 語句，它将任務函數轉變為生成器，來實作一個 "上下文切換"。這個程式使用這個上下文切換來運作任務的兩個執行個體。

程式（<code>example_3.py</code>）的下個版本和上一個版本幾乎完全一樣，除了在我們任務循環體内添加了一個 <code>time.sleep(1)</code> 調用。這使任務循環中的每次疊代都添加了一秒的延遲。這個添加的延遲是為了模拟在我們任務中出現緩慢 IO 操作的影響。

我還導入了一個簡單的 Elapsed Time 類來處理報告中使用的開始時間／已用時間功能。

當該程式運作時，輸出表明任務一和任務二都在運作，消費從隊列裡來的工作并像之前那樣處理它們。随着增加的模拟 IO 操作延遲，我們發現我們協作式的并發并沒有為我們做任何事，延遲會停止整個程式的運作，而 CPU 就隻會等待這個 IO 延遲的結束。

這就是異步文檔中 ”阻塞代碼“的确切含義。注意運作整個程式所需要的時間，你會發現這就是所有 IO 延遲的累積時間。這再次意味着通過這種方式運作程式并不是勝利了。

之後 <code>monkey</code> 子產品一個叫做 <code>patch_all()</code> 的方法被調用。這個方法是用來幹嘛的呢？簡單來說它配置了這個應用程式，使其它所有包含阻塞（同步）代碼的子產品都會被打上"更新檔"，這樣這些同步代碼就會變成異步的。

就像大多數簡單的解釋一樣，這個解釋對你并沒有很大的幫助。在我們示例代碼中與之相關的就是 <code>time.sleep(1)</code>（我們模拟的 IO 延遲）不會再"阻塞"整個程式。取而代之的是它讓出程式的控制傳回給系統。請注意，"example_3.py" 中的 "yield" 語句不再存在，它現在已經是 <code>time.sleep(1)</code> 函數調用内的一部分。

是以，如果 <code>time.sleep(1)</code> 已經被 gevent 打更新檔來讓出控制，那麼這個控制又到哪裡去了？使用 gevent 的一個作用是它會在程式中運作一個事件循環的線程。對于我們的目的來說，這個事件循環就像在 <code>example_3.py</code> 中 "run the tasks" 的循環。當<code>time.sleep(1)</code> 的延遲結束時，它就會把控制傳回給 <code>time.sleep(1)</code> 語句的下一條可執行語句。這樣做的優點是 CPU 不會因為延遲被阻塞，而是可以有空閑去執行其它代碼。

我們 "run the tasks" 的循環已經不再存在了，取而代之的是我們的任務隊列包含了兩個對 <code>gevent.spawn(...)</code> 的調用。這兩個調用會啟動兩個 gevent 線程（叫做 greenlet），它們是互相協作進行上下文切換的輕量級微線程，而不是像普通線程一樣由系統切換上下文。

注意在我們任務生成之後的 <code>gevent.joinall(tasks)</code> 調用。這條語句會讓我們的程式會一直等待任務一和任務二都完成。如果沒有這個的話，我們的程式将會繼續執行後面列印的語句，但是實際上沒有做任何事。

當這個程式運作的時候，請注意任務一和任務二都在同樣的時間開始，然後等待模拟的 IO 調用結束。這表明 <code>time.sleep(1)</code>調用已經不再阻塞，其它的工作也正在被做。

在程式結束時，看下總的運作時間你就會發現它實際上是 <code>example_3.py</code> 運作時間的一半。現在我們開始看到異步程式的優勢了。

在并發運作兩個或者多個事件可以通過非阻塞的方式來執行 IO 操作。通過使用 gevent greenlets 和控制上下文切換，我們就可以在多個任務之間實作多路複用，這個實作并不會遇到太多麻煩。

程式（<code>example_5.py</code>）的下一個版本有一點進步也有一點退步。這個程式現在處理的是有真正 IO 操作的工作，即向一個 URL 清單發起 HTTP 請求來擷取頁面内容，但是它仍然是以阻塞（同步）的方式運作的。

和這個程式之前版本一樣，我們使用一個 <code>yield</code> 關鍵字來把我們的任務函數轉換成生成器，并且為了讓其他任務執行個體可以執行，我們執行了一次上下文切換。

每個任務都會從工作隊列中擷取到一個 URL，擷取這個 URL 指向頁面的内容并且報告擷取這些内容花了多長時間。

和之前一樣，這個 <code>yield</code> 關鍵字讓我們兩個任務都能運作，但是因為這個程式是以同步的方式運作的，每個<code>requests.get()</code> 調用在擷取到頁面之前都會阻塞 CPU。注意在最後運作整個程式的總時間，這對于下一個示例會很有意義。

這個程式（<code>example_6.py</code>）的版本修改了先前的版本再次使用了 gevent 子產品。記得 gevent 子產品的 <code>monkey.patch_all()</code> 調用會修改之後的所有子產品，這樣這些子產品的同步代碼就會變成異步的，其中也包括 <code>requests</code> 子產品。

現在的任務已經改成移除了對 <code>yield</code> 的調用，因為 <code>requests.get(url)</code> 調用已經不會再阻塞了，反而是執行一次上下文切換讓出控制給 gevent 的事件循環。在 “run the task” 部分我們使用 gevent 來産生兩個任務生成器，之後使用 <code>joinall()</code> 來等待它們完成。

在程式運作的最後，你可以看下總共的時間和擷取每個 URL 分别的時間。你将會看到總時間會少于 <code>requests.get()</code> 函數調用的累計時間。

這是因為這些函數調用是異步運作的，是以我們可以同一時間發送多個請求，進而更好地發揮出 CPU的優勢。

Twisted是一個非常強大的系統，采用了和 gevent 根本上不一樣的方式來建立異步程式。gevent 子產品是修改其子產品使它們的同步代碼變成異步，Twisted 提供了它自己的函數和方法來達到同樣的結果。

之前在 <code>example_6.py</code> 中使用被打更新檔的 <code>requests.get(url)</code> 調用來擷取 URL 内容的位置，現在我們使用 Twisted 函數<code>getPage(url)</code>。

在這個版本中，<code>@defer.inlineCallbacks</code> 函數裝飾器和語句 <code>yield getPage(url)</code> 一起實作把上下文切換到 Twisted 的事件循環。

在 gevent 中這個事件循環是隐含的，但是在 Twisted 中，事件循環由位于程式底部的 <code>reactor.run()</code> 明确提供。

注意最後的結果和 gevent 版本一樣，整個程式運作的時間會小于擷取每個 URL 内容的累計時間。

程式 （<code>example_8.py</code>）的這個版本也是使用 Twisted 庫，但是是以更傳統的方式使用 Twisted。

這裡我的意思是不再使用 <code>@defer.inlineCallbacks</code> / <code>yield</code> 這種代碼風格，這個版本會使用明确的回調函數。一個"回調函數"是一個被傳遞給系統的函數，該函數可以在之後的事件響應中被調用。在下面的例子中，<code>success_callback()</code> 被提供給 Twisted，用來在 <code>getPage(url)</code> 調用完成後被調用。

注意在這個程式中 <code>@defer.inlineCallbacks</code> 裝飾器并沒有在 <code>my_task()</code> 函數中使用。除此之外，這個函數産出一個叫做 <code>d</code>的變量，該變量是延後調用的縮寫，是調用函數 <code>getPage(url)</code> 得到的傳回值。

延後是 Twisted 處理異步程式設計的方式，回調函數就附加在其之上。當這個延後"觸發"（即當 <code>getPage(url)</code> 完成時），會以回調函數被附加時定義的變量作為參數，來調用這個回調函數。

運作這個程式的最終結果和先前的兩個示例一樣，運作程式的總時間小于擷取 URLs 内容的總時間。

無論你使用 gevent 還是 Twisted，這隻是個人的喜好和代碼風格問題。這兩個都是強大的庫，提供了讓程式員可以編寫異步代碼的機制。

我希望這可以幫你知道和了解異步程式設計可以在哪裡以及如何可以變得有用。如果你正在編寫一個将 PI 計算到小數點後百萬級别精度的函數，異步代碼對于該程式根本一點用都沒有。

然而，如果你正在嘗試實作一個伺服器，或者是會執行大量 IO 操作的程式，使用異步程式設計就會産生巨大的變化。這是一個強大的技術可以幫助你的程式更上一層樓。

<b>原文釋出時間為：2017年7月20日</b>

<b>本文來自雲栖社群合作夥伴掘金，了解相關資訊可以關注掘金網站。</b>