Python學習之路35-協程

時間 2019-12-05

標籤 python 學習之路欄目 Python 简体版

原文原文鏈接

《流暢的Python》筆記。python

本篇主要討論一個與生成器看似無關，但實際很是相關的概念：協程。微信

1. 前言

說到協程(Coroutine)，若是是剛接觸Python不久的新手，估計第一個反應是：懵逼，這是個什麼玩意兒？有一點基礎的小夥伴可能會想到進程和線程。多線程

其實，和子程序（或者說函數）同樣，協程也是一種程序組件。Donald Knuth曾經說過，子程序是協程的特例。咱們都知道，一個子程序就是一次函數調用，它只有一個入口和一個出口：調用者調用子程序，子程序運行完畢，將結果返回給調用者。而協程則是多入口和多出口的子程序：調用者能夠不止一個，執行過程當中能夠暫停，輸出結果也能夠不止一個。閉包

協程和進程、線程也是有關係的：爲了實現併發，高效利用系統資源，因而有了進程；爲了實現更高的併發，以及減少進程切換時的上下文開銷，因而有了線程；但即使線程切換時的開銷小了，若是線程數量一多(好比10K個)，這時的上下文切換也不可小覷，因而在線程中加入了協程(這裏之因此是「加入」，是由於協程的概念出現得比線程要早)。協程運行在一個線程當中，不會發生線程的切換，而且，它的啓停能夠由用戶自行控制。因爲協程在一個線程中運行，因此在共享資源時不須要加鎖。併發

補充：之後有機會單獨出一篇詳細介紹進程、線程和協程的文章。函數

2. 迭代器、生成器和協程

這三者本不該該放在一塊兒，之因此放在一塊兒，是由於生成器將迭代器和協程聯繫了起來，或者說yield關鍵字將這三者聯繫了起來：生成器能夠做爲迭代器，生成器又是協程比不可少的組成部分。但千萬不要把迭代器用做協程，也別把協程用做迭代器！這二者並不該該存在關係。學習

yield關鍵字背後的機制很強大，它不只能向用戶提供數據，還能從用戶那裏獲取數據。而迭代器、生成器和協程這三個概念實際上是對yield關鍵字用法的取捨：網站

凡是含有關鍵字yield或者yield from的函數都是生成器，無論你是用來幹啥；
若是只是用yield來生成數據，或者說向用戶提供數據，那麼這個生成器能夠看作迭代器(用做迭代器的生成器)；
若是還想用yield來獲取外部的數據，實現雙向數據交換，那麼這個生成器可看作協程(用做協程的生成器)。

這裏先列舉出迭代器和協程在代碼上最直觀的區別：spa

def my_iter(): # 用做迭代器的生成器
    yield 1;  # 做爲迭代器，yield關鍵字後面會跟一個數據
    yield 2;  # 且不關心yield的返回值，沒有賦值語句
    
def my_co(): # 用做協程的生成器
    x = yield    # 這種寫法表示但願從用戶處獲取數據，而不向用戶提供數據(其實提供的是None)
    y = yield 1  # 這種寫法表示既向用戶提供數據，也但願獲得用戶的反饋
複製代碼

3. 協程

本節主要包括協程的運行過程，協程的4個狀態，協程的預激，協程的終止和異常處理，協程的返回值。.net

3.1 協程的運行

協程自己有4個狀態(其實就是生成器的4個狀態)，可使用inspect.getgeneratorstate()函數來肯定：

GEN_CREATED：等待開始執行；
GEN_RUNNING：解釋器正在執行，多線程時能看到這個狀態；
GEN_SUSPENDED：在yield表達式處暫停時的狀態；
GEN_CLOSED：執行結束。

下面經過一個簡單的例子來講明這四個狀態以及協程的運行過程：

>>> def simple_coro(a):
...     print("Started a =", a)
...     b = yield a
...     print("Received b =", b)
...     c = yield a + b
...     print("End with c=", c)
...    
>>> from inspect import getgeneratorstate
>>> my_coro = simple_coro(1)
>>> getgeneratorstate(my_coro)
'GEN_CREATED'       # 剛建立的協程所處的狀態，這時協程尚未被激活
>>> next(my_coro)   ### 第一次調用next()叫作預激，這一步很是重要！ ###
Started a = 1
1
>>> >>> getgeneratorstate(my_coro)
'GEN_SUSPENDED'     # 在yield表達式處暫停時的狀態
>>> my_coro.send(2) # 經過.send()方法將用戶的數據傳給協程
Received b = 2
3
>>> my_coro.send(3)
End with c= 3
Traceback (most recent call last):
  File "<input>", line 1, in <module>
StopIteration       # 協程(生成器)結束，拋出StopIteration
>>> getgeneratorstate(my_coro)
'GEN_CLOSED'        # 協程結束後的狀態
複製代碼

解釋：

剛建立的協程並無激活，對協程的第一次next()調用就是預激，這一步很是重要，它將運行到第一yield表達式處並暫停。對於沒有預激的協程，在調用.send(value)時，若是value不是None，解釋器將拋出異常。對於預激，既能夠調用next()函數，也能夠.send(None)(此時會被特殊處理)。但對於yield from來講則不用預激，它會自動預激。
.send()方法實現了用戶和協程的交互。yield是一個表達式(上述代碼中等號的右邊)，它的默認返回值是None，若是用戶經過.send(value)傳入了參數value，那麼這個值將做爲協程暫停處的yield表達式的返回值。
協程的運行過程：也能夠叫作生成器的運行過程。從上一篇中咱們知道，調用next()函數或.send()方法時，協程會運行到下一個yield表達式處並暫停。具體來講，好比上述代碼中的b = yield a，代碼實際上是停在等號的右邊，yield a這個表達式尚未返回，只是把a傳給了用戶，但尚未計算出yield a表達式的返回值，b所以也沒有被賦值。當代碼再次運行時，等號右邊的yield a表達式才返回值，並將這個值賦給b。若是經過next()函數讓協程繼續運行，則上一個暫停處的**yield表達式將返回默認值**None(b = None)；若是經過.send(value)讓協程繼續運行，則上一個yield表達式將返回value(b = value)。這也解釋了爲何要預激協程：若是沒有預激，也就沒有yield表達式與傳入的value相對應，天然也就拋出異常。

3.2 終止協程和異常處理

協程中沒處理的異常會向上冒泡，傳給next()函數或.send()方法的調用方。不過，咱們也能夠經過.throw()方法手動拋出異常，還能夠經過.close()方法手動結束協程：

generator.throw(exc_type[, exc_value[, traceback]])：讓生成器在暫停的yield表達式處拋出指定的異常。若是生成器處理了這個異常，代碼會向前執行到下一個yield表達式yield a，並將生成的a做爲generator.throw()的返回值。若是生成器沒有處理拋出的異常，則會向上冒泡，而且生成器會終止，狀態轉換成GEN_CLOSED。
generator.close()：使生成器在暫停處的yield表達式處拋出GeneratorExit異常。若是生成器沒有處理這個異常，或者處理時拋出了StopIteration異常，.close()方法直接返回，且不報錯；若是處理GeneratorExit時拋出了非StopIteration異常，則向上冒泡。

3.3 返回值

從上一篇和本篇的代碼中，不知道你們發現了一個現象沒有：全部的生成器最後都沒有寫return語句。這實際上是有緣由的，由於在Python3.3以前，若是生成器返回值，解釋器會報語法錯誤。如今則不會報錯了，但返回的值並非像普通函數那樣能夠直接接收：Python解釋器會把這個返回值綁定到生成器最後拋出的StopIteration異常對象的value屬性中。示例以下：

>>> def test():
...     yield 1
...     return "This is a test"
...
>>> t = test()
>>> next(t)
1
>>> next(t)
Traceback (most recent call last):
  File "<input>", line 1, in <module>
StopIteration: This is a test   # StopIteration有了附加信息
>>> t = test()
>>> next(t)
1
>>> try:
...    next(t)
... except StopIteration as si:
...     print(si.value)  # 獲取返回的值
...
This is a test  
複製代碼

3.4 預激協程的裝飾器

從前文咱們知道，若是要使用協程，必需要預激。能夠手動經過調用next()函數或者.send(None)方法。但有時咱們會忘記手動預激，此時，咱們可使用裝飾器來自動預激協程，這個裝飾器以下：

from functools import wraps

def coroutine(func):
 @wraps(func)
    def primer(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)
        return gen
    return primer
複製代碼

提早預激的生成器只能和yield兼容，不能和yield from兼容，由於yield from會自動預激。因此請肯定你的生成器要不要被放在yield from以後。

4. yield from

上一篇文章說到，對於嵌套生成器，使用yield from能減小不少代碼，好比：

def y2():
    def y1():  # y1只要是個可迭代對象就行
        yield 1
        yield 2
    # 第一種寫法
    for y in y1():
        yield y
    # 第二種寫法
    # yield from y1()

if __name__ == "__main__":
    for y in y2():
        print(y)
複製代碼

第二種寫法明顯比第一種簡潔。這是yield from的一個做用：簡化嵌套循環。yield from後面還能夠跟任意可迭代對象，並非只能跟生成器。

yield from最重要的做用是起到了相似通道的做用：它能讓客戶端代碼和子生成器之間進行數據交換。

這裏有幾個術語須要先解釋一下：

委派生成器：包含yield from <iterable>表達式的生成器函數。
子生成器：上述的<iterable>部分就是子生成器。<iterable>也能夠是委派生成器，以此類推下去，造成一個鏈條，但這個鏈條最終以一個只使用yield表達式的簡單生成器結束。
調用方：調用委派生成器的代碼或對象叫作調用方。爲了不歧異，咱們把最外層的代碼，也就是調用第一層委派生成器的代碼叫作客戶端代碼。

好比上述代碼，按照沒有yield from語句的寫法，若是客戶端代碼想經過y2.send(value)向y1傳值，value只能傳到y2這一層，若是想再傳入y1，將要寫大量複雜的代碼。下面是yield from的說明圖：

結合上圖，可作以下總結：

yield from和yield在使用上並沒有太大區別；
委派生成器也是生成器。當第一次對委派生成器調用next()或.send(None)時，委派生成器會執行到第一個yield from表達式並暫停。當客戶端繼續調用委派生成器的.send()，.throw()和.close()等方法時，會「直接」做用到最內層的子生成器上，而不是讓委派生成器的代碼繼續向前執行。只有當子生成器拋出StopIteration異常後，委派生成器中的代碼才繼續執行，並將StopIteration.value的值做爲yield from表達式的返回值。

補充(可跳過)

這一小節是yield from的邏輯僞代碼實現，代碼較爲複雜，看不懂也沒什麼關係，能夠跳過，也可直接看最後的總結，並不影響yield from的使用。

### "RESULT = yield from EXPR"語句的等效代碼
_i = iter(EXPR)  # 獲得EXPR的迭代器
try:
    _y = next(_i)  # 預激！尚未向客戶端生成值
except StopIteration as _e:  # 若是_i拋出了StopIteration異常
    _r = _e.value  # _i的最後的返回值。這不是最後的生成值！
else:  # 若是調用next(_i)一切正常
    while 1:   # 這是一個無限循環
        try:
            _s = yield _y  # 向客戶端發送子生成器生成的值，而後暫停
        except GeneratorExit as _e:  # 若是客戶端調用.throw(GeneratorExit)，或者調用close方法
            try:  # 首先嚐試獲取_i的close方法，由於_i不必定是生成器，普通迭代器不會實現close方法
                _m = _i.close   
            except AttributeError:
                pass  # 沒有獲取到close方法，什麼也不作
            else:
                _m()  # 若是獲取到了close方法，則調用子生成器的close方法
            raise _e  # 最後無論怎樣，都向上拋出GeneratorExit異常
        except BaseException as _e:  # 若是客戶端經過throw()傳入其它異常
            _x = sys.exc_info()  # 獲取錯誤信息
            try: # 嘗試獲取_i的throw方法，理由和上面的狀況同樣
                _m = _i.throw  
            except AttributeError:  # 若是沒有這個方法
                raise _e            # 則向上拋出用戶傳入的異常
            else:                   # 若是_i有throw方法，即它是一個子生成器
                try:                
                    _y = _m(*_x)    # 嘗試調用子生成器的throw方法
                except StopIteration as _e:
                    _r = _e.value   # 若是子生成器拋出StopIteration，獲取返回的值
                break               # 而且跳出循環
        else:    # 若是在生成器生成值時沒有異常發生
            try: # 試驗證用戶經過.send()方法傳入的值
                if _s is None:  # 若是傳入的是None
                    _y = next(_i)  # 則嘗試調用next()，向前繼續執行
                else:  # 若是傳入的不是None，則嘗試調用子生成器的send方法
                    _y = _i.send(_s)
                    # 若是子生成器沒有send方法，則向上報AttributeError
            except StopIteration as _e: # 若是子生成器拋出了StopIteration
                _r = _e.value           # 獲取子生成器返回的值
                break                   # 並跳出循環，回覆委派生成器的運行
RESULT = _r # _r就是yield from EXPR最終的返回值，將其賦予RESULT
複製代碼

從上面這麼長一串代碼能夠看出，若是沒有yield from，而咱們又想向最內層的子生成器傳值，這得多麻煩。下面總結出幾點yield from的特性：

全部的「直接」其實都是間接的，都是一層一層傳下去，或者一層一層傳上來的，只是咱們感受是直接的而已；
調用.send(value)將值傳給委派生成器時，若是value是None，則調用子生成器的__next__方法；不然，調用子生成器的.send(value)；
當對委派生成器調用.throw()，委派生成器會先肯定子生成器有沒有.throw()方法，若是有，則調用，若是沒有，則向上拋出AttributeError異常；
當客戶端調用委派生成器的.throw(GeneratorExit)或者.close()方法時，委派生成器也會先肯定子生成器有沒有.close()方法，若是有，則調用子生成器的.close()方法，由子生成器來拋出GeneratorExit異常，委派生成器將這個異常向上傳遞；若是子類沒有.close()方法，則委派生成器直接拋出GeneratorExit異常。Python解釋器會捕獲這個異常，但不會顯示異常信息。
只要子生成器拋出StopIteration異常，不論是用戶經過.throw方法傳遞的，仍是子生成器運行結束時拋出的，都會致使委派生成器繼續向前執行。

5. 協程計算均值

在《Python學習之路26》中，咱們分別用類和閉包來實現了平均值的計算，如今，做爲本章最後一個例子，咱們使用協程來實現平均值的計算，其中還會用到yield from和生成器的返回值：

import inspect

def averager():
    total = 0.0
    count = 0
    average = None
    while True:
        term = yield
        if term is None:
            break
        total += term
        count += 1
        average = total / count
    return average

def grouper(results, key):
    while True:  # 每一個循環都會新建averager
        results[key] = yield from averager()

def main(data):
    results = {}
    for key, values in data.items():
        group = grouper(results, key)  # 每一個循環都會新建grouper
        next(group)  # 激活
        for value in values:
            group.send(value)
        # 此句很是重要，不然不會執行到averager()中的return語句，也就得不到最終的返回值
        group.send(None)  

    print(results)

data = {"list1": [1, 2, 3, 4, 5], "list2": [6, 7, 8, 9, 10]}

if __name__ == "__main__":
    main(data)

# 結果：
{'list1': 3.0, 'list2': 8.0}
複製代碼

不知道你們看到這段代碼的時候有沒有什麼疑問。當筆者看到grouper()委派生成器裏的While True:時，很是疑惑：爲啥要加個While循環呢？若是按這個版本，咱們在main中的for循環後檢測group的狀態，會發現它是GEN_SUSPENDED，這筆者的強迫症就犯了，怎麼能不是GEN_CLOSED呢？！並且這個版本每當執行完group.send(None)後，在grouper()中又會建立新的averager，而後當main中group更新後，上一個grouper(也就是剛新建了averager的grouper)因爲引用數爲0，又被回收了。剛新建一個averager就被回收，這很少此一舉嗎？因而筆者將代碼改爲了以下形式：

def grouper(results, key): # 去掉了循環
    results[key] = yield from averager()

def main(data):
    results = {}
    for key, values in data.items():
        -- snip --
        try: # 手動捕獲異常
            group.send(None)
        except StopIteration:
            continue
複製代碼