轉(zhuǎn)載自我自己的 github 博客 ——> 半天鐘的博客
這篇博文講述的 python 協(xié)程是 不正式的、寬泛的協(xié)程 ，即通過客戶調(diào)用 .send(…) 方法發(fā)送數(shù)據(jù)或使用 yield from 結(jié)構(gòu)驅(qū)動(dòng)的生成器函數(shù)， 而不是 asyncio 庫(kù)采用的定義更為嚴(yán)格的協(xié)程。
前言
在 事件驅(qū)動(dòng)型編程 中，協(xié)程常用于離散事件的仿真（在單個(gè)線程中使用一個(gè)主循環(huán)驅(qū)動(dòng)協(xié)程執(zhí)行并發(fā)活動(dòng)）。
協(xié)程通過顯式 自主地把控制權(quán)讓步給中央調(diào)度程序 從而實(shí)現(xiàn)了 協(xié)作式多任務(wù) 。
所以， 協(xié)程是 python 事件驅(qū)動(dòng)型框架和協(xié)作式多任務(wù)的基礎(chǔ)。
那么，弄明白協(xié)程的 進(jìn)化過程 、基本行為和 高效的使用方式 是很有必要的。
本博文想要解釋清楚 python 協(xié)程的基本行為以及如何高效的使用協(xié)程。
在閱讀本文之前，你必須要了解 python 中 yield 關(guān)鍵字、和生成器的基本概念。如果你還不知道這兩個(gè)概念是啥，你可以看我的上一篇博文：淺析 python 迭代器與生成器 或者通過 CSDN 上馮爽朗的博文 簡(jiǎn)單了解 yield 關(guān)鍵字的使用方法。
從生成器到協(xié)程
協(xié)程是指一個(gè)過程，這個(gè)過程與調(diào)用方協(xié)作，即 根據(jù)調(diào)用方提供的值 產(chǎn)出相應(yīng)的值 給調(diào)用方。
從協(xié)程的定義來看，協(xié)程的部分行為和帶有 yield 關(guān)鍵字生成器的行為類似，因?yàn)檎{(diào)用方可以使用 .next() 方法 讓生產(chǎn)器產(chǎn)出值給調(diào)用方。例如，這個(gè)斐波那契生成器函數(shù)：
>>
>
def
fibonacci
(
)
:
a
,
b
=
0
,
1
while
True
:
yield
a
a
,
b
=
b
,
a
+
b
調(diào)用方調(diào)用 next（）函數(shù)可以 獲取它的產(chǎn)出值 ：
>>
>
f
=
fibonacci
(
)
>>
>
print
(
next
(
f
)
)
0
>>
>
print
(
next
(
f
)
)
1
這么看來， 生成器的行為離協(xié)程的行為就差一步，即接收調(diào)用方提供的值。
在 python 2.5 后 yield 關(guān)鍵字就可以在表達(dá)式中使用了，而且生成器 API 中增加了 .send(value)方法。 生成器的調(diào)用方可以使用 .send(…) 方法給生成器發(fā)送數(shù)據(jù)。
這樣一來生成器就可以接收調(diào)用方提供的值了， 其接收的數(shù)據(jù)會(huì)成為 yield 表達(dá)式的值。
例一是一個(gè)簡(jiǎn)單的例子，來說明調(diào)用方如何發(fā)送數(shù)據(jù)及生成器如何接受數(shù)據(jù)。
>>
>
def
coroutine
(
)
:
print
(
'-- 協(xié)程開始 --'
)
x
=
yield
'Nothing'
print
(
'-- 協(xié)程接收到了數(shù)據(jù): {!r} -- '
.
format
(
x
)
)
>>
>
coro
=
coroutine
(
)
<
generator
object
coroutine at
0x10bbb2408
>
>>
>
next
(
coro
)
-
-
協(xié)程開始
-
-
Nothing
>>
>
coro
.
send
(
77
)
-
-
協(xié)程接收到了數(shù)據(jù)
:
77
-
-
Traceback
(
most recent call last
)
:
.
.
.
StopIteration
上面的例子表明：
- 在協(xié)程中， yield 通常出現(xiàn)在表達(dá)式的右邊。
- 調(diào)用方先使用一次 .next() 執(zhí)行 yield ‘Nothing’ 讓協(xié)程產(chǎn)出字符串 “Nothing” 并 懸停至至 yield 表達(dá)式這一行
- 調(diào)用方使用 .send() 發(fā)送數(shù)據(jù)給協(xié)程。
- 發(fā)送的 數(shù)據(jù)代替 yield 表達(dá)式 ，并賦給變量 x。
- 協(xié)程結(jié)束時(shí)與生成器一致，都會(huì)拋出 StopIteration 異常。
需要特別注意的地方有：
首先、調(diào)用方只有在協(xié)程停在了 yield 表達(dá)式時(shí)，才能調(diào)用 .send() 發(fā)送數(shù)據(jù)
，否則，協(xié)程會(huì)拋出 TypeError 異常，如例二：
>>
>
coro
=
coroutine
(
)
>>
>
coro
.
send
(
77
)
Traceback
(
most recent call last
)
:
.
.
.
in
coro
.
send
(
77
)
TypeError
:
can't send non
-
None
value to a just
-
started generator
懸停在 yield 表達(dá)式的協(xié)程狀態(tài)是
GEN_SUSPENDED，你可以使用inspect.getgeneratorstate(...)函數(shù)確定協(xié)程的狀態(tài)。
其次、調(diào)用方使用 .send(y) 發(fā)送的數(shù)據(jù)會(huì)代替協(xié)程中的 yield 表達(dá)式 ，在上例中,發(fā)送的數(shù)據(jù) y 是 77 ,77 代替了 yield 表達(dá)式，并賦給了變量 x。
最后、當(dāng)賦值完畢后、協(xié)程會(huì)繼續(xù)前進(jìn)至下一個(gè) yield 關(guān)鍵字并懸停 ，直至結(jié)束從而拋出 StopIteration 異常。
你可以把 .send( y ) 看做兩個(gè)部分的結(jié)合，即:
- yield 表達(dá)式 = y
- .next()
這樣一來，擁有 .send（）方法的生成器，完全符合了協(xié)程的定義，它可以通 過 .send() 接受調(diào)用方傳遞的值，并且可以通過 yield 產(chǎn)出值給調(diào)用方。
不過，此時(shí)我們沒有辦法在一創(chuàng)建協(xié)程時(shí)，立馬使用它。
你必須要先使用一次 .next() 讓協(xié)程懸停在 yield 表達(dá)式那一行，從而使協(xié)程轉(zhuǎn)變至
GEN_SUSPENDED 狀態(tài)
。這樣的行為被稱作
預(yù)激協(xié)程。
預(yù)激協(xié)程
毫無疑問，預(yù)激協(xié)程是一個(gè)很容易被遺忘的步驟。
需要使用 .send() 發(fā)送數(shù)據(jù)之前還必須使用一次 .next()，這讓人感到厭煩。
我們有什么辦法能夠自動(dòng)預(yù)激協(xié)程呢？
有一種方法是使用能夠提前調(diào)用一次 .next() 的裝飾器，如下面這個(gè) coroutine 裝飾器：
# BEGIN CORO_DECO
>>
>
from
functools
import
wraps
>>
>
def
coroutine_deco
(
func
)
:
"""Decorator: primes `func` by advancing to first `yield`"""
@wraps
(
func
)
#使用 functools.wraps 裝飾器獲得源 func 的所有參數(shù) "*args,**kwargs"
def
primer
(
*
args
,
**
kwargs
)
:
gen
=
func
(
*
args
,
**
kwargs
)
#使用源生成器函數(shù)獲取生成器
next
(
gen
)
#調(diào)用 .next 方法
return
gen
#返回調(diào)用 .next 方法后的生成器
return
primer
# END CORO_DECO
網(wǎng)上有多個(gè)類似的裝飾器。這個(gè)改自 ActiveState 中的一個(gè)訣竅——Pipeline made of coroutines，作者是 Chaobin Tang，而他是受到了 David Beazley 的啟發(fā)。—— 《流暢的 python 》
使用這個(gè)裝飾器后，現(xiàn)在我們?cè)龠\(yùn)行例二的代碼就不會(huì)報(bào) TypeError 異常，而是會(huì)正常運(yùn)行了，如下：
@coroutine_deco
>>
>
def
coroutine
(
)
:
print
(
'-- 協(xié)程開始 --'
)
x
=
yield
'Nothing'
print
(
'協(xié)程接收到了數(shù)據(jù): {!r}'
.
format
(
x
)
)
>>
>
coro
=
coroutine
(
)
-
-
協(xié)程開始
-
-
>>
>
import
inspect
>>
>
inspect
.
getgeneratorstate
(
coro
)
GEN_SUSPENDED
>>
>
cro
.
send
(
77
)
協(xié)程接收到了數(shù)據(jù)
:
77
Traceback
(
most recent call last
)
:
.
.
.
StopIteration
>>
>
inspect
.
getgeneratorstate
(
coro
)
GEN_CLOSED
該例子有如下行為需要注意：
- 在創(chuàng)建協(xié)程 coro 對(duì)象后，直接輸出了 “-- 協(xié)程開始 --” 字符串，這表明， 在創(chuàng)建協(xié)程對(duì)象后，其自動(dòng)調(diào)用了一次 next() 方法。
-
使用
inspect.getgeneratorstate查看協(xié)程的狀態(tài)，發(fā)現(xiàn)其已經(jīng)是GEN_SUSPENDED狀態(tài)， 說明協(xié)程內(nèi)部已經(jīng)懸停在 yield 關(guān)鍵字處。 - 能夠直接調(diào)用 .send() 方法而不用事先使用 .next() 了。
-
協(xié)程結(jié)束時(shí)的狀態(tài)是
GEN_CLOSED
協(xié)程還有一個(gè)很常用的方法 —— .close() 用于提前關(guān)閉協(xié)程。使用該方法后，協(xié)程會(huì)在 yield 表達(dá)式那一行拋出 GeneratorExit 異常。
有時(shí)，我們需要協(xié)程在結(jié)束了所有工作時(shí)，返回一個(gè)值， 這在 python 3.3 之前是不可能的，因?yàn)樵趨f(xié)程的方法體中寫 return 關(guān)鍵字會(huì)報(bào)句法錯(cuò)誤。
讓協(xié)程在終止時(shí)返回值
我們可以在 python 3.3 及之后的版本中 讓終止的協(xié)程返回想要的值 ，只是獲取返回值的方法比較曲折。
下面的例三，定義了一個(gè)動(dòng)態(tài)計(jì)算平均值的協(xié)程，并讓其在結(jié)束工作（接受到 None 值）后 返回一個(gè)元組 ，該元組保存著目前為止收到的數(shù)據(jù)個(gè)數(shù)以及最終的平均值。
>>
>
from
collections
import
namedtuple
>>
>
Result
=
namedtuple
(
'Result'
,
'count average'
)
>>
>
def
averager
(
)
:
total
=
0.0
count
=
0
average
=
None
while
True
:
term
=
yield
average
if
term
is
None
:
break
total
+=
term
count
+=
1
average
=
total
/
count
return
Result
(
count
,
average
)
該函數(shù)有以下行為：
>>
>
coro_avg
=
averager
(
)
>>
>
next
(
coro_avg
)
# <1>
>>
>
coro_avg
.
send
(
10
)
# <2>
10.0
>>
>
coro_avg
.
send
(
30
)
20.0
>>
>
coro_avg
.
send
(
6.5
)
15.5
>>
>
coro_avg
.
send
(
None
)
# <3>
Traceback
(
most recent call last
)
:
.
.
.
StopIteration
:
Result
(
count
=
3
,
average
=
15.5
)
注釋：
① : 手動(dòng)預(yù)激協(xié)程。
② : 調(diào)用 .send(10) 返回目前傳入所有數(shù)的平均值10、之后每傳入一個(gè)數(shù)都能實(shí)時(shí)計(jì)算所有數(shù)的平均值。
③ : 傳入 None ，手動(dòng)結(jié)束該協(xié)程。
注意到，和往常一樣，結(jié)束后 協(xié)程拋出了 StopIteration 異常 。不一樣的是， 該異常保存著返回的值 ，即 Result 對(duì)象。
return 表達(dá)式的值會(huì)偷偷傳給調(diào)用方，賦值給 StopIteration 異常的一個(gè)屬性。這樣做有點(diǎn)不合常理，但是能 保留生成器對(duì)象的常規(guī)行為 ——耗盡時(shí)拋出 StopIteration 異常。
改造上面的代碼，手動(dòng)捕獲異常，獲取返回值，可以這樣寫：
>>
>
coro_avg
=
averager
(
)
>>
>
next
(
coro_avg
)
>>
>
coro_avg
.
send
(
10
)
10.0
>>
>
coro_avg
.
send
(
30
)
20.0
>>
>
coro_avg
.
send
(
6.5
)
15.5
>>
>
try
:
coro_avg
.
send
(
None
)
except
StopIteration
as
exc
:
result
=
exc
.
value
>>
>
result
Result
(
count
=
3
,
average
=
15.5
)
目前，我們說明了如何讓 生成器接收調(diào)用方提供的值從而進(jìn)化成協(xié)程 、如何 使用裝飾器自動(dòng)預(yù)激協(xié)程 、以及 如何從協(xié)程獲取看起來很有用的返回值。
使用協(xié)程似乎太麻煩了點(diǎn) ！
不是嗎？ 為了避免麻煩，我們必須自己定義一個(gè)自動(dòng)預(yù)激協(xié)程的裝飾器，為了獲取協(xié)程的返回值，我們還必須捕捉異常，并獲取異常的 value 屬性。
有什么辦法能夠消除這些麻煩呢?（不用自定義預(yù)激裝飾器也不用捕獲異常以獲得返回值）
在 python 3.3 以后，有一個(gè)新的句法能夠幫助我們解決這些麻煩，即 yield from
yield from 及其工作原理
使用 yield from 關(guān)鍵字 不僅能自動(dòng)預(yù)激協(xié)程 、 自動(dòng)提取異常的 value 屬性返回值作為 yield from 表達(dá)式的值 ，還能夠 作為調(diào)用方和協(xié)程之間的通道 。
如果將例三中的 averager() 改編成使用 yield from 關(guān)鍵字來實(shí)現(xiàn)，會(huì)是例四的代碼：
>>
>
from
collections
import
namedtuple
>>
>
Result
=
namedtuple
(
'Result'
,
'count average'
)
>>
>
def
averager
(
)
:
total
=
0.0
count
=
0
average
=
None
while
True
:
term
=
yield
average
if
term
is
None
:
break
total
+=
term
count
+=
1
average
=
total
/
count
return
Result
(
count
,
average
)
>>
>
result
=
set
(
)
# <1>
>>
>
def
yf_averager
(
result
)
:
# <2>
while
True
:
# <3>
r
=
yield
from
averager
(
)
# <4>
result
.
add
(
r
)
>>
>
yfa
=
yf_averager
(
result
)
# <5>
>>
>
next
(
yfa
)
# <6>
>>
>
yfa
.
send
(
10
)
# <7>
10.0
>>
>
yfa
.
send
(
30
)
20.0
>>
>
yfa
.
send
(
6.5
)
15.5
>>
>
yfa
.
send
(
None
)
# <8>
>>
>
result
# <9>
{
Result
(
count
=
3
,
average
=
15.5
)
}
在例四中，averager() 方法并沒有做任何改變
解釋：
①：創(chuàng)建 result 集合以在調(diào)用方收集結(jié)果。
②：yield from 關(guān)鍵字的
載體函數(shù)
，有時(shí)也叫“委派生成器” ，設(shè)立這一函數(shù)是因?yàn)?
在函數(shù)外部使用 yield from（以及 yield）會(huì)導(dǎo)致句法錯(cuò)誤。
③：使用循環(huán)以保證傳入 None 時(shí)
yf_averager 生成器不拋出 StopIteration 異常
從而直接結(jié)束整個(gè)程序，若是如此，我們便觀察不到 result 了。
④：使用 yield from 關(guān)鍵字后面是
協(xié)程
、前面是接收協(xié)程最終返回值的變量 r，這個(gè) r 我們最終會(huì)放在全局變量 result 集合中。還有一點(diǎn)需要注意、
當(dāng)函數(shù)體重含有 yield from 那么它本身就是協(xié)程了
。
⑤：新建 yf_averager 協(xié)程，以
建立調(diào)用方與 averager 協(xié)程的通道
⑥：預(yù)激 yf_averager 協(xié)程
⑦：使用 .send（）發(fā)送數(shù)據(jù)
⑧：發(fā)送 None 以結(jié)束 averager 協(xié)程
⑨：展示 result 集合中的值，確認(rèn)接收到了最終的結(jié)果
上面如果上面這個(gè)例子你不怎么看得懂，沒關(guān)系，我會(huì)在后面解釋。
你現(xiàn)在
只需要知道 yield from 有這些行為：
- 在例四中，我們沒有預(yù)激 averager 協(xié)程，但是它能夠正常工作。 這說明 yield from 關(guān)鍵字會(huì)自動(dòng)預(yù)激協(xié)程。
- 調(diào)用方使用委派生成器 yf_averager 傳入的值會(huì)送到 averager 里，并且調(diào)用方可以接收到 averager 協(xié)程處理后返回的值。 這說明了使用 yield from 的委派生成器 yf_averager 可以在調(diào)用方和協(xié)程之間建立通道，傳輸數(shù)據(jù)。
- 在獲取 averager 結(jié)果時(shí)，我們沒有捕獲異常，而是在第 22 行代碼中將返回值直接賦給了變量 r。 這說明了協(xié)程的最終返回值會(huì)成為 yield from 表達(dá)式的值。
yield from 關(guān)鍵字的原理
接下來這段偽碼等效于 RESULT = yield from EXPR 語句 。它能夠幫助你理解例四中 yield from 的行為
這并不是完整的偽代碼，它去除了 .throw（）和 .close（）方法，只處理 StopIteration 異常。完整的偽碼在這里 -> yield_from_expansion，不過在理解其功能的方面上，這足夠了。
_i
=
iter
(
EXPR
)
# <1>
try
:
_y
=
next
(
_i
)
# <2>
except
StopIteration
as
_e
:
_r
=
_e
.
value
# <3>
else
:
while
1
:
# <4>
_s
=
yield
_y
# <5>
try
:
_y
=
_i
.
send
(
_s
)
# <6>
except
StopIteration
as
_e
:
# <7>
_r
=
_e
.
value
break
RESULT
=
_r
# <8>
解釋：
① ：EXPR 可以是任何可迭代的對(duì)象，因?yàn)楂@取迭代器 _i（這是子生成器，例子中的 averager 協(xié)程）使用的是 iter() 函數(shù)。
② ：
預(yù)激子生成器（averager 協(xié)程）；結(jié)果保存在 _y 中，作為產(chǎn)出的第一個(gè)值。
③ ：如果拋出 StopIteration 異常，
獲取異常對(duì)象的 value 屬性，賦值給 _r
——這是最簡(jiǎn)單情況下的返回值（RESULT）。
④ ：運(yùn)行這個(gè)循環(huán)時(shí)，委派生成器（yf_averager 生成器）會(huì)阻塞，
只作為調(diào)用方和子生成器之間的通道
。
⑤ ：**產(chǎn)出子生成器當(dāng)前產(chǎn)出的元素；等待調(diào)用方發(fā)送 _s 中保存的值。**因?yàn)檫@一個(gè) yield 表達(dá)式和 ⑥ 中的send()，
委派生成器也變成了協(xié)程。
⑥ ：嘗試讓子生成器向前執(zhí)行，
轉(zhuǎn)發(fā)調(diào)用方發(fā)送的 _s
。
⑦ ：如果子生成器拋出 StopIteration 異常，
獲取 value 屬性的值，賦值給 _r
，然后退出循環(huán)，讓委派生成器恢復(fù)運(yùn)行。
⑧ ：
返回的結(jié)果（RESULT）是 _r
，即整個(gè) yield from 表達(dá)式的值。
以上的偽代碼和注釋，幾戶原封不動(dòng)的搬了《流程的 python 》里的解釋，我只是增加了一些注釋。因?yàn)槲蚁氩怀鋈绾胃玫目偨Y(jié) yield from 關(guān)鍵字的原理。
注意，因?yàn)?yf_averager 是帶 yield 關(guān)鍵字的生成器，所以在 ⑧ 結(jié)束后， 若找不到下一個(gè) yield 關(guān)鍵字，那么 yf_averager 生成器會(huì)拋出 StopIteration 異常 ，這是我在例四中設(shè)立 while 循環(huán) ③ 的直接原因。
我建議你在看懂這段偽代碼的基礎(chǔ)上再去 回顧例四 ，這下你 應(yīng)該豁然開朗 了。如果還看不懂的話，我建議你多花些時(shí)間去看《流程的 python 》的第十六章，該章用了60多頁的篇幅把 python 協(xié)程講得很通透。
結(jié)語
本篇博文中，我用了四個(gè)小節(jié)敘述了我理解中的協(xié)程、及其使用技巧。在一開始，我講述了 協(xié)程是什么 ，及 如何在 python 2.2 及以后的版本中用生成器構(gòu)建協(xié)程 ；然后我講述了 協(xié)程的必要操作（預(yù)激）的自動(dòng)化方法 和 如何在 python 3.3 及以后的版本中獲取協(xié)程的返回值 ；最后，我講述了方便的 yield from 關(guān)鍵字的用法、行為 以及 它的主要原理 。
如果你想要知道 協(xié)程的具體用處 ，《流程的 python 》的第十六章中舉了一個(gè)離散事件仿真的例子—— 出租車隊(duì)運(yùn)營(yíng)仿真 。該仿真程序會(huì)創(chuàng)建幾輛出租車，并模擬他們并行運(yùn)作（離開倉(cāng)庫(kù)、尋找乘客、乘客下次、四處徘徊、回家）。對(duì)于說明如何使用協(xié)程做離散事件仿真是一個(gè)很好的例子。
這是那個(gè)出租車隊(duì)運(yùn)營(yíng)仿真例子的源碼 -> taxi_sim
我希望你看完這篇博文后能夠有所收獲、如果你看到了一些錯(cuò)誤，請(qǐng)?jiān)谠u(píng)論中指出。
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系： 360901061
您的支持是博主寫作最大的動(dòng)力，如果您喜歡我的文章，感覺我的文章對(duì)您有幫助，請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點(diǎn)擊下面給點(diǎn)支持吧，站長(zhǎng)非常感激您！手機(jī)微信長(zhǎng)按不能支付解決辦法：請(qǐng)將微信支付二維碼保存到相冊(cè)，切換到微信，然后點(diǎn)擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
