深入python協程的實現，帶你一層一層揭開協程的神秘面紗！

設計思路

EventLoop和torando的IOLoop一樣，使用IO多路復用來阻塞循環，使用一個Waker來喚醒EventLoop，這點我是照抄的torando，不同之處在於EventLoop中沒有回調函數列表，而是擁有一個協程字典。

協程字典是由一堆協程對象作為key組成的。協程類繼承於Torando::Future，本質上協程對象也是一個Future對象，不過更進一步的是，他同時還是一個生成器代理。key對應的value為每次yield返回的值。

協程類是通過一個裝飾器創建的，代碼如下：

def coroutine(func): def __init__(self, *args, **kwargs): super(self.__class__, self).__init__() self.gen = func(*args, **kwargs) def send(self, result): try: return self.gen.send(result) except (GeneratorExit, StopIteration) as e: self.set_result(e.args[0]) raise def throw(self, typ, val=None, tb=None): return self.gen.throw(typ, val=None, tb=None) def __iter__(self): return self.gen def __await__(self): yield from self cls = type(func.__name__, (Future, Coroutine, Iterable), {"__init__": __init__, "send": send, "throw": throw, "__iter__": __iter__, "__await__": __await__}) return cls

和當IOLoop一樣，EventLoop開始事件循環後，每當有協程對象加入到事件循環中，都會被喚醒，執行循環，循環會處理監聽到的文件描述符，調用其handler，處理timout對象。接下來處理的協程字典中的協程是IOLoop所沒有的操作。

處理協程的方法如下：

def process_coroutine(self, coroutine, yielded): while True: try: # 如果當前yield值為Future對象，則判斷Future是否完成，進而調用send方法髮結果，否則中斷本協程的進行。 # 如果當前yield值不是Future對象, 直接send當前值。 # 如果期間發生協程運行完畢，則刪除該協程。 if isinstance(yielded, Future): if yielded.done(): yielded = self.coroutines[coroutine] = coroutine.send(yielded.result()) else: break else: yielded = self.coroutines[coroutine] = coroutine.send(yielded) except (GeneratorExit, StopIteration): del self.coroutines[coroutine] break

為什麼要這麼實現呢？

假設我們現在擁有一個簡單的協程

demo0

# 通過coroutine裝飾器，該協程會被變成一個協程子類。@coroutinedef normal(): a = yield 1 print("Get a: %s" % a) b = yield 2 print("Get b: %s" % b) c = yield 3 print("Get c: %s" % c) return ccoroutine0 = normal() # 創建一個協程對象loop = EventLoop() # 創建事件循環loop.run_until_complete(coroutine0)# 開啟事件循環直到完成print(coroutine0.result()) # 列印輸出

以上代碼大致會經過如下的操作：

1. 創建一個協程對象
2. 創建事件循環
3. 將協程對象添加到協程字典中
4. 開啟事件循環
5. 事件循環運行到process_coroutine，此時yielded在process_coroutine中由倒數第4行代碼依次變成1， 2， 3(因為協程中的yield返回的不是future類型)，並通過send方法賦值給了a, b, c。直到最後拋出StopIteration。並刪除了該協程
6. 運行結束result方法可以調用了返回return值

通過這個簡單的協程我們大致了解了整個程序的運行流程。但這種協程並沒有什麼卵用，而且把一件簡單的事情弄的非常複雜。接下來再更進一步介紹一個相對複雜的協程。

我們先看一個阻塞函數：

def sleep(seconds): future = Future() def callback(): future.set_result(None) EventLoop().call_at(time.time() + seconds, callback) yield future

這個函數和tornado的sleep函數差不多，函數創建了一個future對象，並在EventLoop中添加了一個Timeout對象，Timeout對象會在指定時間後後被事件循環調用其callback。在這裡，被callback之後，future被set_result，也就是說Future.done()變為了True(記住這一點，後面會用到，很重要)。

下面再寫一個複雜一點的的協程

demo1

@coroutinedef sum(a, b): print("Sum start. %s + %s" % (a, b)) yield from sleep(1) result = yield a + b print("Sum stop. %s + %s" % (a, b)) return resultcoroutine1 = sum(1, 2) # 創建一個協程對象loop = EventLoop() # 創建事件循環loop.run_until_complete(coroutine1)# 開啟事件循環直到完成print(coroutine1.result()) # 列印輸出

以上代碼大致會經過如下的操作：

1. 創建一個協程對象
2. 創建事件循環
3. 將協程對象添加到協程字典中
4. 開啟事件循環
5. 事件循環運行到process_coroutine，sleep返回一個Future對象，並且調用其done方法發現沒有完成。於是跳出循環，回到事件循環。
6. 事件循環會獲取timeouts中所有Timeout對象的最小延遲時長並做為超時時長掛起。這時候整個世界都停止了。
7. 等超時之後，處理timeouts中Timeout對象，發現sleep結束了，調用其callback，future對象這時被置為完成。
8. 事件循環繼續前進到達process_coroutine，此時調用future的done方法發現完成了。send future.result()，繼而執行a + b。隨後的情況和第一個例子無異了。

在這個例子中我們定義了一個sleep子協程，並學習到了如何實現阻塞。

再來看第三個更複雜的例子

demo2

@coroutinedef sum(a, b): print("Sum start. %s + %s" % (a, b)) yield from sleep(1) result = yield a + b print("Sum stop. %s + %s" % (a, b)) return result@coroutinedef multi(a, b): print("Multi start. %s x %s" % (a, b)) yield from sleep(2) result = yield a * b print("Multi stop. %s x %s" % (a, b)) return result@coroutinedef aaddbthenmutilc(a, b, c): sum_result = yield from sum(a, b) multi_result = yield from multi(sum_result, c) return multi_resultcoroutine2 = aaddbthenmutilc() # 創建一個協程對象loop = EventLoop() # 創建事件循環loop.run_until_complete(coroutine2)# 開啟事件循環直到完成print(coroutine2.result()) # 列印輸出

剛才說了協程也是一個future，但協程本身還是一個生成器。所以當我們返回協程時，使用yield from 代替yield。

當使用yield from 時生成器和子生器之間就建立起了一個長長的管道，子生器生成的值可以直接返回，無論鏈接多少個生成器，就好像直接調用一個生成器一樣。。

在這裡coroutine2中調用了sum子協程。一但其結束，sum的值會被賦值給sum_result，coroutine2繼續執行multi，直到結束。

寫到現在，可能有同學會問了，如果是真正的IO阻塞，怎麼處理。比如socket。別急，下面馬上給出第4個例子。

在此之前，我又實現了一個類似於sleep的阻塞函數

def get_buffer(socket): future = Future() def callback(fd_obj, events): future.set_result(fd_obj.recv(1024)) EventLoop().add_handler(socket, callback, EventLoop.READ) buffer = yield future return buffer

這個函數是專門用來處理socket讀產生的阻塞的，創建一個future，並創建一個回調函數，添加到事件循環中，EventLoop中有一組對象被稱為handlers，是用來處理文件描述符的。這個和tornado的IOLoop保持一致。通過add_handler可以添加。

在這裡，socket被監聽了READ事件。

demo3

@coroutinedef socket_coroutine(): print("socket_coroutine start. ") import socket from toolkit.event_loop import get_buffer client = socket.socket() client.connect(("", 1234)) buffer = yield from get_buffer(client) client.close() print("socket_coroutine stop. ") return buffercoroutine3 = socket_coroutine() # 創建一個協程對象loop = EventLoop() # 創建事件循環loop.run_until_complete(coroutine3)# 開啟事件循環直到完成print(coroutine3.result()) # 列印輸出

同時我開啟一個server

In [3]: def serve(): ...: server = socket.socket() ...: server.bind(("", 1234)) ...: server.listen(0) ...: client, addr = server.accept() ...: import time ...: time.sleep(5) ...: client.send(b"hello world") ...: client.close() ...: server.close() ...:In [4]: serve()

這個例子和sleep差不多，不同之處在於sleep讓事件循環停止了一段時間，而通過IO多路復用則是在被監聽的文件描述符發生讀事件時激活了事件循環。之後的步驟和demo1一樣。

下面還有最後一個例子，就是多個協程一起運行。

demo4

不貼代碼了，太長，代碼在此！https://github.com/ShichaoMa/toolkit/blob/master/test/test_event_loop.py

裡面有6個協程。

總結

其實關鍵的思路就在於如何處理協程生成器上，我的想法是對於一個協程，無非就是存在兩種情況，阻塞與不阻塞，阻塞的情況是返回的是future同時done為False,這種情況就像sleep ，get_buffer一樣。剛開始返回的都是沒有完成的future，導致該生成器無法被繼續執行，這個時候process_coroutine方法是會直接停止該生成器的執行，繼續執行下一個生成器，生成timeouts對象和添加事件監聽保證了在規定時間內（timout.callback）或者在事件發生時future被置為done=True。對於沒有阻塞的情況（返回的是非Future或者Future已經完成），生成器會一直執行下去直到結束。

終於寫完了。

希望能給您帶一些編程靈感！這個事件循環的所有代碼都被放在我的開發工具包toolkit裡面了。可在通過 pip install toolkity==1.6.0安裝

ShichaoMa/toolkitgithub.com

如果您喜歡同時方便的話，請Start一下，感謝關注。

推薦閱讀：