文章目錄

• • 1. 協(xié)程工作流程和狀態(tài)
  • 2. 預(yù)激協(xié)程的裝飾器
  • 3. 終止協(xié)程、異常處理
  • 4. 讓協(xié)程返回值
  • 5. yield from

learn from 《流暢的python》

1. 協(xié)程工作流程和狀態(tài)

def simple_coroutine(): # 協(xié)程使用生成器函數(shù)定義，有yield關(guān)鍵字print("-> coroutine started")x = yield # yield 右邊沒有表達(dá)式，所以只需從客戶那里接收數(shù)據(jù)print("-> coroutine received: ", x)my_coro = simple_coroutine() # 調(diào)用函數(shù)，得到生成器對(duì)象 print(my_coro) # <generator object simple_coroutine at 0x00000192FD458E40> print(next(my_coro)) # 調(diào)用next 到 yield 處暫停 # -> coroutine started # None my_coro.send(24) # 調(diào)用send，yield 會(huì)計(jì)算出24，之后協(xié)程恢復(fù)， # 一直運(yùn)行到下一個(gè) yield 表達(dá)式，或者到達(dá)終止迭代 # -> 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

可以查看協(xié)程的狀態(tài) print(inspect.getgeneratorstate((my_coro)))，4種狀態(tài)

• GEN_CREATED 等待開始
• GEN_RUNNING 正在執(zhí)行（多線程中可見）
• GEN_SUSPENDED 在 yield 表達(dá)式處暫停
• GEN_CLOSED 執(zhí)行結(jié)束

啟動(dòng)協(xié)程：

• 調(diào)用 next()
• 調(diào)用 obj.send(None)，必須是 None，否則報(bào)錯(cuò)

def simple_coroutine2(a):print("-> started: a = ", a)b = yield a # = 號(hào)右邊的代碼先執(zhí)行，并暫停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

計(jì)算平均數(shù)的例子：

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. 預(yù)激協(xié)程的裝飾器

使用 next， 或者 send（None）

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

• 用 yield from 句法調(diào)用協(xié)程時(shí)，會(huì)自動(dòng)預(yù)激
• asyncio.coroutine 裝飾器不會(huì)預(yù)激協(xié)程，因此 能兼容 yield from 句法

3. 終止協(xié)程、異常處理

協(xié)程中未處理的異常會(huì)向上冒泡，傳給 next 函數(shù)或 send 方法的調(diào)用方（即觸發(fā)協(xié)程的對(duì)象）。

• 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.")# 上面處理了異常，這一行永遠(yuǎn)不會(huì)被執(zhí)行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

• 處理了異常，協(xié)程可以繼續(xù)執(zhí)行

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

• 沒有處理異常，發(fā)生異常，協(xié)程終止

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

• 如果不管協(xié)程如何結(jié)束都想做些清理工作，要把協(xié)程定義體中相關(guān)的代碼放入 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. 讓協(xié)程返回值

from collections import namedtuple res = namedtuple("Result", "count average")def averager():tot = 0.0count = 0avg = Nonewhile True:term = yieldif term is None:break # 為了返回值，協(xié)程必須正常終止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) # 發(fā)送None終止循環(huán)，協(xié)程結(jié)束 # 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)

• 如何獲取協(xié)程的返回值，捕獲異常，讀取異常的 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

在其他語言 中，類似的結(jié)構(gòu)使用 await 關(guān)鍵字

• 在生成器 gen 中使用 yield from subgen() 時(shí)，subgen 會(huì)獲得控制權(quán)，把產(chǎn)出的值傳給 gen 的調(diào)用方，即調(diào)用方 可以直接控制 subgen
• 與此同時(shí)，gen 會(huì)阻塞，等待 subgen 終止

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

yield from x :

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

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

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

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。