文章目錄

• • 1. 協程工作流程和狀態
  • 2. 預激協程的裝飾器
  • 3. 終止協程、異常處理
  • 4. 讓協程返回值
  • 5. yield from

learn from 《流暢的python》

1. 協程工作流程和狀態

def simple_coroutine(): # 協程使用生成器函數定義，有yield關鍵字print("-> coroutine started")x = yield # yield 右邊沒有表達式，所以只需從客戶那里接收數據print("-> coroutine received: ", x)my_coro = simple_coroutine() # 調用函數，得到生成器對象 print(my_coro) # <generator object simple_coroutine at 0x00000192FD458E40> print(next(my_coro)) # 調用next 到 yield 處暫停 # -> coroutine started # None my_coro.send(24) # 調用send，yield 會計算出24，之后協程恢復， # 一直運行到下一個 yield 表達式，或者到達終止迭代 # -> coroutine received: 24 # Traceback (most recent call last): # File "D:/gitcode/Python_learning/fluent_python/coroutine.py", line 12, in <module> # my_coro.send(24) # StopIteration

可以查看協程的狀態 print(inspect.getgeneratorstate((my_coro)))，4種狀態

• GEN_CREATED 等待開始
• GEN_RUNNING 正在執行（多線程中可見）
• GEN_SUSPENDED 在 yield 表達式處暫停
• GEN_CLOSED 執行結束

啟動協程：

• 調用 next()
• 調用 obj.send(None)，必須是 None，否則報錯

def simple_coroutine2(a):print("-> started: a = ", a)b = yield a # = 號右邊的代碼先執行，并暫停print("-> Received: b = ", b)c = yield a + bprint("-> Received: c = ", c)my_coro2 = simple_coroutine2(24) print(inspect.getgeneratorstate((my_coro2))) # GEN_CREATED print(next(my_coro2)) # -> started: a = 24 # 24 print(inspect.getgeneratorstate((my_coro2))) # GEN_SUSPENDED print(my_coro2.send(6)) # -> Received: b = 6 # 30 try:print(my_coro2.send(3)) # -> Received: c = 3 except:pass print(inspect.getgeneratorstate((my_coro2))) # GEN_CLOSED

計算平均數的例子：

def avg():tot = 0.0count = 0average = Nonewhile True:term = yield averagetot += termcount += 1average = tot / counta = avg() a.send(None) print(a.send(2)) # 2.0 print(a.send(3)) # 2.5 print(a.close()) # None print(a.send(3)) # StopIteration

2. 預激協程的裝飾器

使用 next， 或者 send（None）

from functools import wrapsdef coroutine(func):@wraps(func)def primer(*args, **kwargs): # 被裝飾的生成器函數替換成 primer函數gen = func(*args, **kwargs) # 調用被裝飾的函數，獲取生成器對象next(gen) # 預激生成器return gen # 返回生成器return primer

• 用 yield from 句法調用協程時，會自動預激
• asyncio.coroutine 裝飾器不會預激協程，因此 能兼容 yield from 句法

3. 終止協程、異常處理

協程中未處理的異常會向上冒泡，傳給 next 函數或 send 方法的調用方（即觸發協程的對象）。

• generator.throw(exc_type[, exc_value[, traceback]])
• generator.close()

class DemoException(Exception):passdef demo_exc_handling():print("-> coroutine started")while True:try:x = yieldexcept DemoException:print("*** DemoException handled. continuing")else:print("-> coroutine received: {!r}".format(x))raise RuntimeError("This line should never run.")# 上面處理了異常，這一行永遠不會被執行exc_coro = demo_exc_handling() next(exc_coro) # -> coroutine started exc_coro.send(11) # -> coroutine received: 11 exc_coro.send(12) # -> coroutine received: 12 exc_coro.close() print(inspect.getgeneratorstate(exc_coro)) # GEN_CLOSED

• 處理了異常，協程可以繼續執行

exc_coro = demo_exc_handling() next(exc_coro) # -> coroutine started exc_coro.send(11) # -> coroutine received: 11 exc_coro.throw(DemoException) # *** DemoException handled. continuing print(inspect.getgeneratorstate(exc_coro)) # GEN_SUSPENDED

• 沒有處理異常，發生異常，協程終止

exc_coro = demo_exc_handling() next(exc_coro) # -> coroutine started exc_coro.send(11) # -> coroutine received: 11 exc_coro.throw(ZeroDivisionError) # Traceback (most recent call last): # File "D:/gitcode/Python_learning/fluent_python/coroutine.py", line 116, in <module> # exc_coro.throw(ZeroDivisionError) # File "D:/gitcode/Python_learning/fluent_python/coroutine.py", line 92, in demo_exc_handling # x = yield # ZeroDivisionError

• 如果不管協程如何結束都想做些清理工作，要把協程定義體中相關的代碼放入 try/finally 塊中

print("--------------") def demo_finally():print("-> coroutine started")try:while True:try:x = yieldexcept DemoException:print("*** DemoException handled. continuing")else:print("-> coroutine received: {!r}".format(x))finally:print("-> coroutine ending")exc_coro = demo_finally() next(exc_coro) exc_coro.send(10) exc_coro.throw(DemoException) exc_coro.send(12) # -> coroutine started # -> coroutine received: 10 # *** DemoException handled. continuing # -> coroutine received: 12 # -> coroutine ending

4. 讓協程返回值

from collections import namedtuple res = namedtuple("Result", "count average")def averager():tot = 0.0count = 0avg = Nonewhile True:term = yieldif term is None:break # 為了返回值，協程必須正常終止tot += termcount += 1avg = tot/countreturn res(count, avg) coro_avg = averager() next(coro_avg) coro_avg.send(10) coro_avg.send(20) coro_avg.send(30) coro_avg.send(None) # 發送None終止循環，協程結束 # Traceback (most recent call last): # File "D:/gitcode/Python_learning/fluent_python/coroutine.py", line 170, in <module> # coro_avg.send(None) # StopIteration: Result(count=3, average=20.0)

• 如何獲取協程的返回值，捕獲異常，讀取異常的 value

coro_avg = averager() next(coro_avg) coro_avg.send(10) coro_avg.send(20) coro_avg.send(30) try:coro_avg.send(None) except StopIteration as exc:result = exc.value print(result) # Result(count=3, average=20.0)

5. yield from

在其他語言 中，類似的結構使用 await 關鍵字

• 在生成器 gen 中使用 yield from subgen() 時，subgen 會獲得控制權，把產出的值傳給 gen 的調用方，即調用方 可以直接控制 subgen
• 與此同時，gen 會阻塞，等待 subgen 終止

def gen():yield from "AB"yield from range(1,3) print(list(gen())) # ['A', 'B', 1, 2]

yield from x :

• 首先調用 iter(x)，獲取迭代器

yield from 的主要功能是打開雙向通道，把最外層的調用方與最內層 的子生成器連接起來，這樣二者可以直接發送和產出值，還可以直接傳入異常，而不用在位于中間的協程中添加大量處理異常的樣板代碼。
有了這個結構，協程可以通過 以前不可能的方式委托職責

上面圖中左側 外層 for 循環的末尾沒有 group.send(None)，則子生成器不終止，委派生成器 會在 yield from 處永遠暫停

• 還可以用協程做 離散事件仿真
• 如果想使用現成的 Python 協程庫，可以使用 SimPy

總結

以上是生活随笔為你收集整理的python 协程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。