[进阶] --- Python3 异步编程详解（史上最全篇）
?
[進階] - Python3 異步編程詳解：https://blog.csdn.net/lu8000/article/details/45025987
參考：http://aosabook.org/en/500L/a-web-crawler-with-asyncio-coroutines.html
木風卜雨：https://blog.csdn.net/lu8000
?
?
1 什么是異步編程
?
1.1 阻塞
- 程序未得到所需計算資源時被掛起的狀態。
- 程序在等待某個操作完成期間，自身無法繼續干別的事情，則稱該程序在該操作上是阻塞的。
- 常見的阻塞形式有：網絡I/O阻塞、磁盤I/O阻塞、用戶輸入阻塞等。
阻塞是無處不在的，包括CPU切換上下文時，所有的進程都無法真正干事情，它們也會被阻塞。（如果是多核CPU則正在執行上下文切換操作的核不可被利用。）
?
1.2 非阻塞
- 程序在等待某操作過程中，自身不被阻塞，可以繼續運行干別的事情，則稱該程序在該操作上是非阻塞的。
- 非阻塞并不是在任何程序級別、任何情況下都可以存在的。
- 僅當程序封裝的級別可以囊括獨立的子程序單元時，它才可能存在非阻塞狀態。
非阻塞的存在是因為阻塞存在，正因為某個操作阻塞導致的耗時與效率低下，我們才要把它變成非阻塞的。
?
1.3 同步
- 不同程序單元為了完成某個任務，在執行過程中需靠某種通信方式以協調一致，稱這些程序單元是同步執行的。
- 例如購物系統中更新商品庫存，需要用“行鎖”作為通信信號，讓不同的更新請求強制排隊順序執行，那更新庫存的操作是同步的。
- 簡言之，同步意味著有序。
?
1.4 異步
- 為完成某個任務，不同程序單元之間過程中無需通信協調，也能完成任務的方式。
- 不相關的程序單元之間可以是異步的。
- 例如，爬蟲下載網頁。調度程序調用下載程序后，即可調度其他任務，而無需與該下載任務保持通信以協調行為。不同網頁的下載、保存等操作都是無關的，也無需相互通知協調。這些異步操作的完成時刻并不確定。
- 簡言之，異步意味著無序。
上文提到的“通信方式”通常是指異步和并發編程提供的同步原語，如信號量、鎖、同步隊列等等。我們需知道，雖然這些通信方式是為了讓多個程序在一定條件下同步執行，但正因為是異步的存在，才需要這些通信方式。如果所有程序都是按序執行，其本身就是同步的，又何需這些同步信號呢？
?
1.5 并發
- 并發描述的是程序的組織結構。指程序要被設計成多個可獨立執行的子任務。
- 以利用有限的計算機資源使多個任務可以被實時或近實時執行為目的。
?
1.6 并行
- 并行描述的是程序的執行狀態。指多個任務同時被執行。
- 以利用富余計算資源（多核CPU）加速完成多個任務為目的。
并發提供了一種程序組織結構方式，讓問題的解決方案可以并行執行，但并行執行不是必須的。
?
1.7 概念總結
- 并行是為了利用多核加速多任務完成的進度
- 并發是為了讓獨立的子任務都有機會被盡快執行，但不一定能加速整體進度
- 非阻塞是為了提高程序整體執行效率
- 異步是高效地組織非阻塞任務的方式
要支持并發，必須拆分為多任務，不同任務相對而言才有阻塞/非阻塞、同步/異步。所以，并發、異步、非阻塞三個詞總是如影隨形。
?
1.8 異步編程
- 以進程、線程、協程、函數/方法作為執行任務程序的基本單位，結合回調、事件循環、信號量等機制，以提高程序整體執行效率和并發能力的編程方式。
如果在某程序的運行時，能根據已經執行的指令準確判斷它接下來要進行哪個具體操作，那它是同步程序，反之則為異步程序。（無序與有序的區別）
同步/異步、阻塞/非阻塞并非水火不容，要看討論的程序所處的封裝級別。例如購物程序在處理多個用戶的瀏覽請求可以是異步的，而更新庫存時必須是同步的。
?
1.9 異步之難(nán)
- 控制不住 “計幾” 寫的程序，因為其執行順序不可預料，當下正要發生什么事件不可預料。在并行情況下更為復雜和艱難。所以，幾乎所有的異步框架都將異步編程模型簡化：一次只允許處理一個事件。故而有關異步的討論幾乎都集中在了單線程內。
- 如果某事件處理程序需要長時間執行，所有其他部分都會被阻塞。所以，一旦采取異步編程，每個異步調用必須“足夠小”，不能耗時太久。如何拆分異步任務成了難題。
- 程序下一步行為往往依賴上一步執行結果，如何知曉上次異步調用已完成并獲取結果？
- 回調（Callback）成了必然選擇。那又需要面臨“回調地獄”的折磨。
- 同步代碼改為異步代碼，必然破壞代碼結構。
- 解決問題的邏輯也要轉變，不再是一條路走到黑，需要精心安排異步任務。
?
?
2 苦心異步為哪般
?
如上文所述，異步編程面臨諸多難點，Python 之父親自上陣打磨4年才使 asyncio 模塊在Python 3.6中“轉正”，如此苦心為什么？答案只有一個：它值得！下面我們看看為何而值得。
?
2.1 CPU的時間觀
我們將一個 2.6GHz 的 CPU 擬人化，假設它執行一條命令的時間，他它感覺上過了一秒鐘。CPU是計算機的處理核心，也是最寶貴的資源，如果有浪費CPU的運行時間，導致其利用率不足，那程序效率必然低下（因為實際上有資源可以使效率更高）。
如上圖所示，在千兆網上傳輸2KB數據，CPU感覺過了14個小時，如果是在10M的公網上呢？那效率會低百倍！如果在這么長的一段時間內，CPU只是傻等結果而不能去干其他事情，是不是在浪費CPU的青春？
魯迅說，浪費“CPU”的時間等于謀財害命。而兇手就是程序猿。
?
2.2 面臨的問題
- 成本問題
如果一個程序不能有效利用一臺計算機資源，那必然需要更多的計算機通過運行更多的程序實例來彌補需求缺口。例如我前不久主導重寫的項目，使用Python異步編程，改版后由原來的7臺服務器削減至3臺，成本驟降57%。一臺AWS m4.xlarge 型通用服務器按需付費實例一年價格約 1.2 萬人民幣。
- 效率問題
如果不在乎錢的消耗，那也會在意效率問題。當服務器數量堆疊到一定規模后，如果不改進軟件架構和實現，加機器是徒勞，而且運維成本會驟然增加。比如別人家的電商平臺支持6000單/秒支付，而自家在下單量才支撐2000單/秒，在雙十一這種活動的時候，錢送上門也賺不到。
- C10k/C10M挑戰
C10k（concurrently handling 10k connections）是一個在1999年被提出來的技術挑戰，如何在一顆1GHz CPU，2G內存，1gbps網絡環境下，讓單臺服務器同時為1萬個客戶端提供FTP服務。而到了2010年后，隨著硬件技術的發展，這個問題被延伸為C10M，即如何利用8核心CPU，64G內存，在10gbps的網絡上保持1000萬并發連接，或是每秒鐘處理100萬的連接。（兩種類型的計算機資源在各自的時代都約為1200美元）
成本和效率問題是從企業經營角度講，C10k/C10M問題則是從技術角度出發挑戰軟硬件極限。C10k/C10M 問題得解，成本問題和效率問題迎刃而解。
?
2.3 解決方案
《約束理論與企業優化》中指出：“除了瓶頸之外，任何改進都是幻覺。”
CPU告訴我們，它自己很快，而上下文切換慢、內存讀數據慢、磁盤尋址與取數據慢、網絡傳輸慢……總之，離開CPU 后的一切，除了一級高速緩存，都很慢。我們觀察計算機的組成可以知道，主要由運算器、控制器、存儲器、輸入設備、輸出設備五部分組成。運算器和控制器主要集成在CPU中，除此之外全是I/O，包括讀寫內存、讀寫磁盤、讀寫網卡全都是I/O。I/O成了最大的瓶頸。
異步程序可以提高效率，而最大的瓶頸在I/O，業界誕生的解決方案沒出意料：異步I/O吧，異步I/O吧，異步I/O吧吧！
?
?
3 異步I/O進化之路
?
如今，地球上最發達、規模最龐大的計算機程序，莫過于因特網。而從CPU的時間觀中可知，網絡I/O是最大的I/O瓶頸，除了宕機沒有比它更慢的。所以，諸多異步框架都對準的是網絡I/O。
我們從一個爬蟲例子說起，從因特網上下載10篇網頁。
?
3.1 同步阻塞方式
最容易想到的解決方案就是依次下載，從建立socket連接到發送網絡請求再到讀取響應數據，順序進行。
import socket import datetimedef blocking_way():sock = socket.socket()# blockingsock.connect(('example.com', 80))request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'sock.send(request.encode('ascii'))response = b''chunk = sock.recv(4096)while chunk:response += chunk# blockingchunk = sock.recv(4096)return responsedef sync_way():start_time = datetime.datetime.now()res = []for i in range(10):res.append(blocking_way())end_time = datetime.datetime.now()print(f'Cost time:{(end_time - start_time).seconds}')return len(res)if __name__ == '__main__':print(sync_way())注：總體耗時約為4.5秒。（因網絡波動每次測試結果有所變動，本文取多次平均值）
如上圖所示，blocking_way()?的作用是建立 socket 連接，發送HTTP請求，然后從 socket讀取HTTP響應并返回數據。示例中我們請求了 example.com 的首頁。在sync_way()?執行了10次，即下載 example.com 首頁10次。
在示例代碼中有兩個關鍵點。一是第10行的** sock.connect((‘example.com’, 80))，該調用的作用是向example.com主機的80端口發起網絡連接請求。 二是第14行、第18行的sock.recv(4096)**，該調用的作用是從socket上讀取4K字節數據。
我們知道，創建網絡連接，多久能創建完成不是客戶端決定的，而是由網絡狀況和服務端處理能力共同決定。服務端什么時候返回了響應數據并被客戶端接收到可供程序讀取，也是不可預測的。所以sock.connect()和sock.recv()這兩個調用在默認情況下是阻塞的。
注：sock.send()函數并不會阻塞太久，它只負責將請求數據拷貝到TCP/IP協議棧的系統緩沖區中就返回，并不等待服務端返回的應答確認。
假設網絡環境很差，創建網絡連接需要1秒鐘，那么sock.connect()就得阻塞1秒鐘，等待網絡連接成功。這1秒鐘對一顆2.6GHz的CPU來講，仿佛過去了83年，然而它不能干任何事情。sock.recv()也是一樣的必須得等到服務端的響應數據已經被客戶端接收。我們下載10篇網頁，這個阻塞過程就得重復10次。如果一個爬蟲系統每天要下載1000萬篇網頁呢？！
上面說了很多，我們力圖說明一件事：同步阻塞的網絡交互方式，效率低十分低下。特別是在網絡交互頻繁的程序中。這種方式根本不可能挑戰C10K/C10M。
?
3.2 改進方式：多進程
在一個程序內，依次執行10次太耗時，那開10個一樣的程序同時執行不就行了。于是我們想到了多進程編程。為什么會先想到多進程呢？發展脈絡如此。在更早的操作系統（Linux 2.4）及其以前，進程是 OS 調度任務的實體，是面向進程設計的OS。
import socket import datetime from concurrent import futuresdef blocking_way():sock = socket.socket()# blockingsock.connect(('example.com', 80))request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'sock.send(request.encode('ascii'))response = b''chunk = sock.recv(4096)while chunk:response += chunk# blockingchunk = sock.recv(4096)return responsedef process_way():start_time = datetime.datetime.now()workers = 10with futures.ProcessPoolExecutor(workers) as executor:futs = {executor.submit(blocking_way) for i in range(10)}end_time = datetime.datetime.now()print(f'Cost time:{(end_time - start_time).seconds}')return len([fut.result() for fut in futs])if __name__ == '__main__':print(process_way())注：總體耗時約為 0.6 秒。
改善效果立竿見影。但仍然有問題?？傮w耗時并沒有縮減到原來的十分之一，而是九分之一左右，還有一些時間耗到哪里去了？進程切換開銷。
進程切換開銷不止像“CPU的時間觀”所列的“上下文切換”那么低。CPU從一個進程切換到另一個進程，需要把舊進程運行時的寄存器狀態、內存狀態全部保存好，再將另一個進程之前保存的數據恢復。對CPU來講，幾個小時就干等著。當進程數量大于CPU核心數量時，進程切換是必然需要的。
除了切換開銷，多進程還有另外的缺點。一般的服務器在能夠穩定運行的前提下，可以同時處理的進程數在數十個到數百個規模。如果進程數量規模更大，系統運行將不穩定，而且可用內存資源往往也會不足。
多進程解決方案在面臨每天需要成百上千萬次下載任務的爬蟲系統，或者需要同時搞定數萬并發的電商系統來說，并不適合。
除了切換開銷大，以及可支持的任務規模小之外，多進程還有其他缺點，如狀態共享等問題，后文會有提及，此處不再細究。
?
3.3 繼續改進：多線程
由于線程的數據結構比進程更輕量級，同一個進程可以容納多個線程，從進程到線程的優化由此展開。后來的OS也把調度單位由進程轉為線程，進程只作為線程的容器，用于管理進程所需的資源。而且OS級別的線程是可以被分配到不同的CPU核心同時運行的。
import socket import datetime from concurrent import futuresdef blocking_way():sock = socket.socket()# blockingsock.connect(('example.com', 80))request = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'sock.send(request.encode('ascii'))response = b''chunk = sock.recv(4096)while chunk:response += chunk# blockingchunk = sock.recv(4096)return responsedef process_way():start_time = datetime.datetime.now()workers = 10with futures.ThreadPoolExecutor(workers) as executor:futs = {executor.submit(blocking_way) for i in range(10)}end_time = datetime.datetime.now()print(f'Cost time:{(end_time - start_time).seconds}')return len([fut.result() for fut in futs])if __name__ == '__main__':print(process_way())注：總體運行時間約0.43秒。
結果符合預期，比多進程耗時要少些。從運行時間上看，多線程似乎已經解決了切換開銷大的問題。而且可支持的任務數量規模，也變成了數百個到數千個。
但是，多線程仍有問題，特別是Python里的多線程。首先，Python中的多線程因為GIL的存在，它們并不能利用CPU多核優勢，一個Python進程中，只允許有一個線程處于運行狀態。那為什么結果還是如預期，耗時縮減到了十分之一？
因為在做阻塞的系統調用時，例如sock.connect(),sock.recv()時，當前線程會釋放GIL，讓別的線程有執行機會。但是單個線程內，在阻塞調用上還是阻塞的。
小提示：Python中 time.sleep 是阻塞的，都知道使用它要謹慎，但在多線程編程中，time.sleep 并不會阻塞其他線程。
除了GIL之外，所有的多線程還有通病。它們是被OS調度，調度策略是搶占式的，以保證同等優先級的線程都有均等的執行機會，那帶來的問題是：并不知道下一時刻是哪個線程被運行，也不知道它正要執行的代碼是什么。所以就可能存在競態條件。
例如爬蟲工作線程從任務隊列拿待抓取URL的時候，如果多個爬蟲線程同時來取，那這個任務到底該給誰？那就需要用到“鎖”或“同步隊列”來保證下載任務不會被重復執行。
而且線程支持的多任務規模，在數百到數千的數量規模。在大規模的高頻網絡交互系統中，仍然有些吃力。當然，多線程最主要的問題還是競態條件。
?
3.4 非阻塞方式
終于，我們來到了非阻塞解決方案。先來看看最原始的非阻塞如何工作的。
import socket import datetimedef nonblocking_way():sock = socket.socket()sock.setblocking(False)try:sock.connect(('example.com', 80))except BlockingIOError:# 非阻塞連接過程中也會拋出異常passrequest = 'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'data = request.encode('ascii')# 不知道 sock 何時就緒，所以不斷嘗試連接while True:try:sock.send(data)breakexcept OSError:passresponse = b''while True:try:chunk = sock.recv(4096)while chunk:response += chunkchunk = sock.recv(4096)breakexcept OSError:passreturn responsedef sync_way():start_time = datetime.datetime.now()res = []for i in range(10):res.append(nonblocking_way())end_time = datetime.datetime.now()print(f'Cost time:{(end_time - start_time).seconds}')return len(res)if __name__ == '__main__':print(sync_way())注：總體耗時約4.3秒。
首先注意到兩點，就感覺被騙了。一是耗時與同步阻塞相當，二是代碼更復雜。要非阻塞何用？且慢。
上圖第9行代碼sock.setblocking(False)告訴OS，讓socket上阻塞調用都改為非阻塞的方式。之前我們說到，非阻塞就是在做一件事的時候，不阻礙調用它的程序做別的事情。上述代碼在執行完 **sock.connect() **和?sock.recv()?后的確不再阻塞，可以繼續往下執行請求準備的代碼或者是執行下一次讀取。
代碼變得更復雜也是上述原因所致。第11行要放在try語句內，是因為socket在發送非阻塞連接請求過程中，系統底層也會拋出異常。connect()被調用之后，立即可以往下執行第15和16行的代碼。
需要while循環不斷嘗試?send()，是因為connect()已經非阻塞，在send()之時并不知道?socket?的連接是否就緒，只有不斷嘗試，嘗試成功為止，即發送數據成功了。recv()調用也是同理。
雖然 connect() 和 recv() 不再阻塞主程序，空出來的時間段CPU沒有空閑著，但并沒有利用好這空閑去做其他有意義的事情，而是在循環嘗試讀寫 socket （不停判斷非阻塞調用的狀態是否就緒）。還得處理來自底層的可忽略的異常。也不能同時處理多個 socket 。
然后10次下載任務仍然按序進行。所以總體執行時間和同步阻塞相當。如果非得這樣子，那還不如同步阻塞算了。
?
3.5 非阻塞改進
?
3.5.1 epoll
判斷非阻塞調用是否就緒如果 OS 能做，是不是應用程序就可以不用自己去等待和判斷了，就可以利用這個空閑去做其他事情以提高效率。
所以OS將I/O狀態的變化都封裝成了事件，如可讀事件、可寫事件。并且提供了專門的系統模塊讓應用程序可以接收事件通知。這個模塊就是select。讓應用程序可以通過select注冊文件描述符和回調函數。當文件描述符的狀態發生變化時，select?就調用事先注冊的回調函數。
select因其算法效率比較低，后來改進成了poll，再后來又有進一步改進，BSD內核改進成了kqueue模塊，而Linux內核改進成了epoll模塊。這四個模塊的作用都相同，暴露給程序員使用的API也幾乎一致，區別在于kqueue?和?epoll?在處理大量文件描述符時效率更高。
鑒于 Linux 服務器的普遍性，以及為了追求更高效率，所以我們常常聽聞被探討的模塊都是?epoll?。
?
3.5.2 回調(Callback)
把I/O事件的等待和監聽任務交給了 OS，那 OS 在知道I/O狀態發生改變后（例如socket連接已建立成功可發送數據），它又怎么知道接下來該干嘛呢？只能回調。
需要我們將發送數據與讀取數據封裝成獨立的函數，讓epoll代替應用程序監聽socket狀態時，得告訴epoll：“如果socket狀態變為可以往里寫數據（連接建立成功了），請調用HTTP請求發送函數。如果socket?變為可以讀數據了（客戶端已收到響應），請調用響應處理函數。”
于是我們利用epoll結合回調機制重構爬蟲代碼：
import socket import datetime from selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselector = DefaultSelector() stopped = False urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}class Crawler(object):def __init__(self, url=None):self.url = urlself.sock = Noneself.response = b''def fetch(self):self.sock = socket.socket()self.sock.setblocking(False)try:self.sock.connect(('example.com', 80))except BlockingIOError:passselector.register(self.sock.fileno(), EVENT_WRITE, self.connected)def connected(self, key, mask):selector.unregister(key.fd)get = f'GET {self.url} HTTP/1.0\r\nHost: example.com\r\n\r\n'self.sock.send(get.encode('ascii'))selector.register(key.fd, EVENT_READ, self.read_response)def read_response(self, key, mask):global stopped# 如果相應大于 4KB ，下一次循環會繼續讀取chunk = self.sock.recv(4096)if chunk:self.response += chunkelse:selector.unregister(key.fd)urls_todo.remove(self.url)if not urls_todo:stopped = True此處和前面稍有不同的是，我們將下載不同的10個頁面，相對URL路徑存放于urls_todo集合中?，F在看看改進在哪。
首先，不斷嘗試send()?和?recv()?的兩個循環被消滅掉了。
其次，導入了selectors模塊，并創建了一個DefaultSelector?實例。Python標準庫提供的selectors模塊是對底層select/poll/epoll/kqueue的封裝。DefaultSelector類會根據 OS 環境自動選擇最佳的模塊，那在 Linux 2.5.44 及更新的版本上都是epoll了。
然后，在第25行和第31行分別注冊了socket可寫事件(EVENT_WRITE)和可讀事件(EVENT_READ)發生后應該采取的回調函數。
雖然代碼結構清晰了，阻塞操作也交給OS去等待和通知了，但是，我們要抓取10個不同頁面，就得創建10個Crawler實例，就有20個事件將要發生，那如何從selector里獲取當前正發生的事件，并且得到對應的回調函數去執行呢？
?
3.5.3 事件循環（Event Loop）
為了解決上述問題，那我們只得采用老辦法，寫一個循環，去訪問selector模塊，等待它告訴我們當前是哪個事件發生了，應該對應哪個回調。這個等待事件通知的循環，稱之為事件循環。
def loop():while not stopped:# 阻塞，直到一個事件發生events = selector.select()for event_key, event_mask in events:callback = event_key.datacallback(event_key, event_mask)上述代碼中，我們用stopped全局變量控制事件循環何時停止。當urls_todo消耗完畢后，會標記stopped為True。
重要的是第49行代碼，selector.select() 是一個阻塞調用，因為如果事件不發生，那應用程序就沒事件可處理，所以就干脆阻塞在這里等待事件發生。那可以推斷，如果只下載一篇網頁，一定要connect()之后才能send()繼而recv()，那它的效率和阻塞的方式是一樣的。因為不在connect()/recv()上阻塞，也得在select()上阻塞。
所以，selector機制(后文以此稱呼代指epoll/kqueue)是設計用來解決大量并發連接的。當系統中有大量非阻塞調用，能隨時產生事件的時候，selector機制才能發揮最大的威力。
下面是如何啟創建10個下載任務和啟動事件循環的：
import socket from selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselector = DefaultSelector() stopped = False urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}class Crawler(object):def __init__(self, url=None):self.url = urlself.sock = Noneself.response = b''def fetch(self):self.sock = socket.socket()self.sock.setblocking(False)try:self.sock.connect(('example.com', 80))except BlockingIOError:passselector.register(self.sock.fileno(), EVENT_WRITE, self.connected)def connected(self, key, mask):selector.unregister(key.fd)get = f'GET {self.url} HTTP/1.0\r\nHost: example.com\r\n\r\n'self.sock.send(get.encode('ascii'))selector.register(key.fd, EVENT_READ, self.read_response)def read_response(self, key, mask):global stopped# 如果相應大于 4KB ，下一次循環會繼續讀取chunk = self.sock.recv(4096)if chunk:self.response += chunkelse:selector.unregister(key.fd)urls_todo.remove(self.url)if not urls_todo:stopped = Truedef loop():while not stopped:# 阻塞，直到一個事件發生events = selector.select()for event_key, event_mask in events:callback = event_key.datacallback(event_key, event_mask)if __name__ == '__main__':import timestart = time.time()for url in urls_todo:crawler = Crawler(url)crawler.fetch()loop()print(time.time() - start)注：總體耗時約0.45秒。
上述執行結果令人振奮。在單線程內用?事件循環+回調?搞定了10篇網頁同時下載的問題。這，已經是異步編程了。雖然有一個for 循環順序地創建Crawler 實例并調用** fetch** 方法，但是fetch?內僅有connect()和注冊可寫事件，而且從執行時間明顯可以推斷，多個下載任務確實在同時進行！
上述代碼異步執行的過程：
?
3.5.4 總結
目前為止，我們已經從同步阻塞學習到了異步非阻塞。掌握了在單線程內同時并發執行多個網絡I/O阻塞型任務的黑魔法。而且與多線程相比，連線程切換都沒有了，執行回調函數是函數調用開銷，在線程的棧內完成，因此性能也更好，單機支持的任務規模也變成了數萬到數十萬個。（不過我們知道：沒有免費午餐，也沒有銀彈。）
部分編程語言中，對異步編程的支持就止步于此（不含語言官方之外的擴展）。需要程序猿直接使用epoll去注冊事件和回調、維護一個事件循環，然后大多數時間都花在設計回調函數上。
通過本節的學習，我們應該認識到，不論什么編程語言，但凡要做異步編程，上述的“事件循環+回調”這種模式是逃不掉的，盡管它可能用的不是epoll，也可能不是while循環。如果你找到了一種不屬于 “等會兒告訴你” 模型的異步方式，請立即給我打電話（注意，打電話是Call）。
為什么我們在某些異步編程中并沒有看到 CallBack 模式呢？這就是我們接下來要探討的問題。本節是學習異步編程的一個終點，也是另一個起點。畢竟咱們講 Python 異步編程，還沒提到其主角協程的用武之地。
?
?
4 Python 對異步I/O的優化之路
?
我們將在本節學習到 Python 生態對異步編程的支持是如何繼承前文所述的“事件循環+回調”模式演變到 asyncio 的原生協程模式。
?
4.1 回調之痛，以終為始
在第3節中，我們已經學會了“事件循環+回調”的基本運行原理，可以基于這種方式在單線程內實現異步編程。也確實能夠大大提高程序運行效率。但是，剛才所學的只是最基本的，然而在生產項目中，要應對的復雜度會大大增加?？紤]如下問題：
- 如果回調函數執行不正常該如何？
- 如果回調里面還要嵌套回調怎么辦？要嵌套很多層怎么辦？
- 如果嵌套了多層，其中某個環節出錯了會造成什么后果？
- 如果有個數據需要被每個回調都處理怎么辦？
- ……
在實際編程中，上述系列問題不可避免。在這些問題的背后隱藏著回調編程模式的一些缺點：
- 回調層次過多時代碼可讀性差
- 破壞代碼結構
寫同步代碼時，關聯的操作時自上而下運行：
do_a() do_b()如果 b 處理依賴于 a 處理的結果，而 a 過程是異步調用，就不知 a 何時能返回值，需要將后續的處理過程以callback的方式傳遞給 a ，讓 a 執行完以后可以執行 b。代碼變化為：
do_a(do_b())如果整個流程中全部改為異步處理，而流程比較長的話，代碼邏輯就會成為這樣：
do_a(do_b(do_c(do_d(do_e(do_f(......))))))上面實際也是回調地獄式的風格，但這不是主要矛盾。主要在于，原本從上而下的代碼結構，要改成從內到外的。先f，再e，再d，…，直到最外層 a 執行完成。在同步版本中，執行完a后執行b，這是線程的指令指針控制著的流程，而在回調版本中，流程就是程序猿需要注意和安排的。
- 共享狀態管理困難
回顧第3節爬蟲代碼，同步阻塞版的sock對象從頭使用到尾，而在回調的版本中，我們必須在Crawler實例化后的對象self里保存它自己的sock對象。如果不是采用OOP的編程風格，那需要把要共享的狀態接力似的傳遞給每一個回調。多個異步調用之間，到底要共享哪些狀態，事先就得考慮清楚，精心設計。 - 錯誤處理困難
一連串的回調構成一個完整的調用鏈。例如上述的 a 到 f。假如 d 拋了異常怎么辦？整個調用鏈斷掉，接力傳遞的狀態也會丟失，這種現象稱為調用棧撕裂。 c 不知道該干嘛，繼續異常，然后是 b 異常，接著 a 異常。好嘛，報錯日志就告訴你，a 調用出錯了，但實際是 d 出錯。所以，為了防止棧撕裂，異常必須以數據的形式返回，而不是直接拋出異常，然后每個回調中需要檢查上次調用的返回值，以防錯誤吞沒。
如果說代碼風格難看是小事，但棧撕裂和狀態管理困難這兩個缺點會讓基于回調的異步編程很艱難。所以不同編程語言的生態都在致力于解決這個問題。才誕生了后來的Promise、Co-routine等解決方案。
Python 生態也以終為始，秉承著“程序猿不必難程序猿”的原則，讓語言和框架開發者苦逼一點，也要讓應用開發者舒坦。在事件循環+回調的基礎上衍生出了基于協程的解決方案，代表作有?Tornado、Twisted、asyncio?等。接下來我們隨著 Python 生態異步編程的發展過程，深入理解Python異步編程。
?
4.2 核心問題
通過前面的學習，我們清楚地認識到異步編程最大的困難：異步任務何時執行完畢？接下來要對異步調用的返回結果做什么操作？
上述問題我們已經通過事件循環和回調解決了。但是回調會讓程序變得復雜。要異步，必回調，又是否有辦法規避其缺點呢？那需要弄清楚其本質，為什么回調是必須的？還有使用回調時克服的那些缺點又是為了什么？
答案是程序為了知道自己已經干了什么？正在干什么？將來要干什么？換言之，程序得知道當前所處的狀態，而且要將這個狀態在不同的回調之間延續下去。
多個回調之間的狀態管理困難，那讓每個回調都能管理自己的狀態怎么樣？鏈式調用會有棧撕裂的困難，讓回調之間不再鏈式調用怎樣？不鏈式調用的話，那又如何讓被調用者知道已經完成了？那就讓這個回調通知那個回調如何？而且一個回調，不就是一個待處理任務嗎？
任務之間得相互通知，每個任務得有自己的狀態。那不就是很古老的編程技法：協作式多任務？然而要在單線程內做調度，啊哈，協程！每個協程具有自己的棧幀，當然能知道自己處于什么狀態，協程之間可以協作那自然可以通知別的協程。
?
4.3 協程
- 協程(Co-routine)，即是協作式的例程。
它是非搶占式的多任務子例程的概括，可以允許有多個入口點在例程中確定的位置來控制程序的暫停與恢復執行。
例程是什么？編程語言定義的可被調用的代碼段，為了完成某個特定功能而封裝在一起的一系列指令。一般的編程語言都用稱為函數或方法的代碼結構來體現。
?
4.4 基于生成器的協程
早期的 Pythoner 發現 Python 中有種特殊的對象——生成器（Generator），它的特點和協程很像。每一次迭代之間，會暫停執行，繼續下一次迭代的時候還不會丟失先前的狀態。
為了支持用生成器做簡單的協程，Python 2.5 對生成器進行了增強(PEP 342)，該增強提案的標題是 “Coroutines via Enhanced Generators”。有了PEP 342的加持，生成器可以通過?yield?暫停執行和向外返回數據，也可以通過?send()?向生成器內發送數據，還可以通過?throw()?向生成器內拋出異常以便隨時終止生成器的運行。
接下來，我們用基于生成器的協程來重構先前的爬蟲代碼。
?
4.4.1 未來對象(Future)
不用回調的方式了，怎么知道異步調用的結果呢？先設計一個對象，異步調用執行完的時候，就把結果放在它里面。這種對象稱之為未來對象。
import socket from selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselector = DefaultSelector() stopped = False urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}class Future(object):def __init__(self):self.result = Noneself._callbacks = []def add_done_callback(self, fn):self._callbacks.append(fn)def set_result(self, result=None):self.result = resultfor fn in self._callbacks:fn(self)未來對象有一個?result?屬性，用于存放未來的執行結果。還有個?set_result()?方法，是用于設置?result?的，并且會在給?result?綁定值以后運行事先給 future 添加的回調?；卣{是通過未來對象的add_done_callback()?方法添加的。
不要疑惑此處的?callback，說好了不回調的嘛？難道忘了我們曾經說的要異步，必回調。不過也別急，此處的回調，和先前學到的回調，還真有點不一樣。
?
4.4.2 重構 Crawler
現在不論如何，我們有了未來對象可以代表未來的值。先用?Future?來重構爬蟲代碼
import socket from selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselector = DefaultSelector() stopped = False urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}class Future(object):def __init__(self):self.result = Noneself._callbacks = []def add_done_callback(self, fn):self._callbacks.append(fn)def set_result(self, result=None):self.result = resultfor fn in self._callbacks:fn(self)class Crawler(object):def __init__(self, url=None):self.url = urlself.response = b''def fetch(self):sock = socket.socket()sock.setblocking(False)try:sock.connect(('example.com', 80))except BlockingIOError:passf = Future()def on_connected():f.set_result(None)selector.register(sock.fileno(), EVENT_WRITE, on_connected)yield fselector.unregister(sock.fileno())get = f'GET {self.url} HTTP/1.0\r\nHost: example.com\r\n\r\n'sock.send(get.encode('ascii'))global stoppedwhile True:f = Future()def on_readable():f.set_result(sock.recv(4096))selector.register(sock.fileno(), EVENT_READ, on_readable())chunk = yield fselector.unregister(sock.fileno())if chunk:self.response += chunkelse:urls_todo.remove(self.url)if not urls_todo:stopped = Truebreak和先前的回調版本對比，已經有了較大差異。fetch 方法內有了 yield 表達式，使它成為了生成器。我們知道生成器需要先調用?next()?迭代一次或者是先?send(None)?啟動，遇到 yield 之后便暫停。
那?fetch 生成器如何再次恢復執行呢？至少?Future?和?Crawler都沒看到相關代碼。。。 繼續往下看。。。
?
4.4.3 任務對象(Task)
為了解決上述問題，我們只需遵循一個編程規則：單一職責，每種角色各司其職，如果還有工作沒有角色來做，那就創建一個角色去做。沒人來恢復這個生成器的執行么？沒人來管理生成器的狀態么？創建一個，就叫?Task?好了，很合適的名字。
class Task(object):def __init__(self, t_coroutine=None):self.t_coroutine = t_coroutinet_f = Future()t_f.set_result(None)self.step(t_f)def step(self, t_future=None):try:# send 會進入到 coroutine 執行，即 fetch，直到下次 yield# next_future 為 yield 返回的對象next_future = self.coroutine.send(t_future.result)except StopIteration:returnnext_future.add_done_callback(self.step)上述代碼中?Task?封裝了?t_coroutine?對象，即初始化時傳遞給他的對象，被管理的任務是待執行的協程，故而這里的 t_coroutine 就是?fetch()生成器。它還有個step()方法，在初始化的時候就會執行一遍。step()?內會調用生成器的send()方法，初始化第一次發送的是None就驅動了 t_coroutine 即?fetch()?的第一次執行。
send()完成之后，得到下一次的?future，然后給下一次的?future?添加?step()?回調。原來?add_done_callback()?不是給寫爬蟲業務邏輯用的。此前的?callback?可就干的是業務邏輯呀。
再看?fetch()?生成器，其內部寫完了所有的業務邏輯，包括如何發送請求，如何讀取響應。而且注冊給 selector 的回調相當簡單，就是給對應的?future?對象綁定結果值。兩個 yield 表達式都是返回對應的 future對象，然后返回?Task.step()?之內，這樣?Task, Future, Coroutine?三者精妙地串聯在了一起。
初始化 Task 對象以后，把?fetch()?給驅動到了第44行?yied f?就完事了，接下來怎么繼續？
?
4.4.4 事件循環(Event Loop)驅動協程運行
該事件循環上場了。接下來，只需等待已經注冊的 EVENT_WRITE 事件發生。事件循環就像心臟一般，只要它開始跳動，整個程序就會持續運行。
注：總體耗時約0.43秒。
現在loop有了些許變化，callback()不再傳遞event_key和event_mask參數。也就是說，這里的回調根本不關心是誰觸發了這個事件，結合fetch()可以知道，它只需完成對future設置結果值即可f.set_result()。而且future是誰它也不關心，因為協程能夠保存自己的狀態，知道自己的future是哪個。也不用關心到底要設置什么值，因為要設置什么值也是協程內安排的。
此時的loop()，真的成了一個心臟，它只管往外泵血，不論這份血液是要輸送給大腦還是要給腳趾，只要它還在跳動，生命就能延續。
?
4.4.5 生成器協程風格和回調風格對比總結
在回調風格中：
- 存在鏈式回調（雖然示例中嵌套回調只有一層）
- 請求和響應也不得不分為兩個回調以至于破壞了同步代碼那種結構
- 程序員必須在回調之間維護必須的狀態。
還有更多示例中沒有展示，但確實存在的問題，參見4.1節。
而基于生成器協程的風格：
- 無鏈式調用
- selector的回調里只管給future設置值，不再關心業務邏輯
- loop 內回調callback()不再關注是誰觸發了事件
- 已趨近于同步代碼的結構
- 無需程序員在多個協程之間維護狀態，例如哪個才是自己的sock
?
4.4.6 碉堡了，但是代碼很丑！能不能重構？
如果說fetch的容錯能力要更強，業務功能也需要更完善，怎么辦？而且技術處理的部分（socket相關的）和業務處理的部分（請求與返回數據的處理）混在一起。
- 創建socket連接可以抽象復用吧？
- 循環讀取整個response可以抽象復用吧？
- 循環內處理socket.recv()的可以抽象復用吧？
但是這些關鍵節點的地方都有yield，抽離出來的代碼也需要是生成器。而且fetch()自己也得是生成器。生成器里玩生成器，代碼好像要寫得更丑才可以……
Python 語言的設計者們也認識到了這個問題，再次秉承著“程序猿不必為難程序猿”的原則，他們搗鼓出了一個yield from來解決生成器里玩生成器的問題。
?
4.5 用 yield from 改進生成器協程
?
4.5.1 yield from語法介紹
yield from 是Python 3.3 新引入的語法（PEP 380）。它主要解決的就是在生成器里玩生成器不方便的問題。它有兩大主要功能。
第一個功能是：讓嵌套生成器不必通過循環迭代yield，而是直接yield from。以下兩種在生成器里玩子生成器的方式是等價的。
def gen_one():subgen = range(10) yield from subgendef gen_two():subgen = range(10) for item in subgen: yield item第二個功能就是在子生成器和原生成器的調用者之間打開雙向通道，兩者可以直接通信。
def gen():yield from subgen()def subgen():while True:x = yieldyield x+1def main():g = gen()next(g) # 驅動生成器g開始執行到第一個 yieldretval = g.send(1) # 看似向生成器 gen() 發送數據print(retval) # 返回2g.throw(StopIteration) # 看似向gen()拋入異常通過上述代碼清晰地理解了yield from的雙向通道功能。關鍵字yield from在gen()內部為subgen()和main()開辟了通信通道。main()里可以直接將數據1發送給subgen(),subgen()也可以將計算后的數據2返回到main()里，main()里也可以直接向subgen()拋入異常以終止subgen()。
順帶一提，yield from 除了可以 yield from 還可以 yield from 。
?
4.5.2 重構代碼
抽象socket連接的功能:
抽象單次?recv()?和讀取完整的 response 功能:
三個關鍵點的抽象已經完成，現在重構Crawler類：
上面代碼整體來講沒什么問題，可復用的代碼已經抽象出去，作為子生成器也可以使用?yield from?語法來獲取值。但另外有個點需要注意：在第24和第35行返回future對象的時候，我們了yield from f?而不是原來的yield f。yield可以直接作用于普通Python對象，而yield from卻不行，所以我們對Future還要進一步改造，把它變成一個iterable對象就可以了。
只是增加了iter()方法的實現。如果不把Future改成iterable也是可以的，還是用原來的yield?f即可。那為什么需要改進呢？
首先，我們是在基于生成器做協程，而生成器還得是生成器，如果繼續混用yield和yield from?做協程，代碼可讀性和可理解性都不好。其次，如果不改，協程內還得關心它等待的對象是否可被yield，如果協程里還想繼續返回協程怎么辦？如果想調用普通函數動態生成一個Future對象再返回怎么辦？
所以，在Python 3.3 引入yield from新語法之后，就不再推薦用yield去做協程。全都使用yield from由于其雙向通道的功能，可以讓我們在協程間隨心所欲地傳遞數據。
?
4.5.3 yield from改進協程總結
用yield from改進基于生成器的協程，代碼抽象程度更高。使業務邏輯相關的代碼更精簡。由于其雙向通道功能可以讓協程之間隨心所欲傳遞數據，使Python異步編程的協程解決方案大大向前邁進了一步。
于是Python語言開發者們充分利用yield from，使?Guido?主導的Python異步編程框架Tulip迅速脫胎換骨，并迫不及待得讓它在 Python 3.4 中換了個名字asyncio以“實習生”角色出現在標準庫中。
?
4.5.4 asyncio 介紹
asyncio是Python 3.4 試驗性引入的異步I/O框架（PEP 3156），提供了基于協程做異步I/O編寫單線程并發代碼的基礎設施。其核心組件有事件循環（Event Loop）、協程(Coroutine）、任務(Task)、未來對象(Future)以及其他一些擴充和輔助性質的模塊。
在引入asyncio的時候，還提供了一個裝飾器@asyncio.coroutine用于裝飾使用了yield from的函數，以標記其為協程。但并不強制使用這個裝飾器。
雖然發展到 Python 3.4 時有了yield from的加持讓協程更容易了，但是由于協程在Python中發展的歷史包袱所致，很多人仍然弄不明白生成器和協程的聯系與區別，也弄不明白yield和?yield from?的區別。這種混亂的狀態也違背Python之禪的一些準則。
于是Python設計者們又快馬加鞭地在 3.5 中新增了async/await語法（PEP 492），對協程有了明確而顯式的支持，稱之為原生協程。async/await?和?yield from這兩種風格的協程底層復用共同的實現，而且相互兼容。
在Python 3.6 中asyncio庫“轉正”，不再是實驗性質的，成為標準庫的正式一員。
?
4.6 總結
行至此處，我們已經掌握了asyncio的核心原理，學習了它的原型，也學習了異步I/O在 CPython 官方支持的生態下是如何一步步發展至今的。
實際上，真正的asyncio比我們前幾節中學到的要復雜得多，它還實現了零拷貝、公平調度、異常處理、任務狀態管理等等使 Python 異步編程更完善的內容。理解原理和原型對我們后續學習有莫大的幫助。
?
?
5 asyncio和原生協程初體驗
?
本節中，我們將初步體驗?asyncio庫和新增語法?async/await?給我們帶來的便利。由于Python2-3的過度期間，Python3.0-3.4的使用者并不是太多，也為了不讓更多的人困惑，也因為?aysncio?在 3.6 才轉正，所以更深入學習 asyncio 庫的時候我們將使用?async/await?定義的原生協程風格，yield from?風格的協程不再闡述（實際上它們可用很小的代價相互代替）。
import asyncio import aiohttphost = 'http://example.com' urls_todo = {'/', '/1', '/2', '/3', '/4', '/5', '/6', '/7', '/8', '/9'}loop = asyncio.get_event_loop()async def fetch(url=None):async with aiohttp.ClientSession(loop=loop) as session:async with session.get(url=url) as response:res = await response.read()return resif __name__ == '__main__':import timestart = time.time()tasks = [fetch(host + url) for url in urls_todo]loop.run_until_complete(asyncio.gather(*tasks))print(time.time() - start)pass對比生成器版的協程，使用asyncio庫后變化很大：
- 沒有了yield?或?yield from，而是async/await
- 沒有了自造的loop()，取而代之的是asyncio.get_event_loop()
- 無需自己在socket上做異步操作，不用顯式地注冊和注銷事件，aiohttp庫已經代勞
- 沒有了顯式的** Future** 和** Task，asyncio**已封裝
- 更少量的代碼，更優雅的設計
說明：我們這里發送和接收HTTP請求不再自己操作socket的原因是，在實際做業務項目的過程中，要處理妥善地HTTP協議會很復雜，我們需要的是功能完善的異步HTTP客戶端，業界已經有了成熟的解決方案，DRY不是嗎？
和同步阻塞版的代碼對比：
- 異步化
- 代碼量相當（引入aiohttp框架后更少）
- 代碼邏輯同樣簡單，跟同步代碼一樣的結構、一樣的邏輯
- 接近10倍的性能提升
?
?
結語
?
到此為止，我們已經深入地學習了異步編程是什么、為什么、在Python里是怎么樣發展的。我們找到了一種讓代碼看起來跟同步代碼一樣簡單，而效率卻提升N倍（具體提升情況取決于項目規模、網絡環境、實現細節）的異步編程方法。它也沒有回調的那些缺點。
本系列教程接下來的一篇將是學習asyncio庫如何的使用，快速掌握它的主要內容。后續我們還會深入探究asyncio的優點與缺點，也會探討Python生態中其他異步I/O方案和asyncio的區別。
?
?
?
?
總結
以上是生活随笔為你收集整理的[进阶] --- Python3 异步编程详解（史上最全篇）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 创建型、结构型、行为型模式（2）
- 下一篇： Java8 Stream详解~遍历/匹配