理解Python的協(xié)程(Coroutine)

由于GIL的存在，導(dǎo)致Python多線程性能甚至比單線程更糟。

GIL: 全局解釋器鎖（英語(yǔ)：Global Interpreter Lock，縮寫(xiě)GIL），是計(jì)算機(jī)程序設(shè)計(jì)語(yǔ)言解釋器用于同步線程的一種機(jī)制，它使得任何時(shí)刻僅有一個(gè)線程在執(zhí)行。[1]即便在多核心處理器上，使用 GIL 的解釋器也只允許同一時(shí)間執(zhí)行一個(gè)線程。

于是出現(xiàn)了協(xié)程（Coroutine）這么個(gè)東西。

協(xié)程: 協(xié)程，又稱微線程，纖程，英文名Coroutine。協(xié)程的作用，是在執(zhí)行函數(shù)A時(shí)，可以隨時(shí)中斷，去執(zhí)行函數(shù)B，然后中斷繼續(xù)執(zhí)行函數(shù)A（可以自由切換）。但這一過(guò)程并不是函數(shù)調(diào)用（沒(méi)有調(diào)用語(yǔ)句），這一整個(gè)過(guò)程看似像多線程，然而協(xié)程只有一個(gè)線程執(zhí)行.

協(xié)程由于由程序主動(dòng)控制切換，沒(méi)有線程切換的開(kāi)銷，所以執(zhí)行效率極高。對(duì)于IO密集型任務(wù)非常適用，如果是cpu密集型，推薦多進(jìn)程+協(xié)程的方式。

在Python3.4之前，官方?jīng)]有對(duì)協(xié)程的支持，存在一些三方庫(kù)的實(shí)現(xiàn)，比如gevent和Tornado。3.4之后就內(nèi)置了asyncio標(biāo)準(zhǔn)庫(kù)，官方真正實(shí)現(xiàn)了協(xié)程這一特性。

而Python對(duì)協(xié)程的支持，是通過(guò)Generator實(shí)現(xiàn)的，協(xié)程是遵循某些規(guī)則的生成器。因此，我們?cè)诹私鈪f(xié)程之前，我們先要學(xué)習(xí)生成器。

生成器(Generator)

我們這里主要討論yield和yield from這兩個(gè)表達(dá)式，這兩個(gè)表達(dá)式和協(xié)程的實(shí)現(xiàn)息息相關(guān)。

  • Python2.5中引入yield表達(dá)式，參見(jiàn)PEP342
  • Python3.3中增加yield from語(yǔ)法，參見(jiàn)PEP380，

方法中包含yield表達(dá)式后，Python會(huì)將其視作generator對(duì)象，不再是普通的方法。

yield表達(dá)式的使用

我們先來(lái)看該表達(dá)式的具體使用：

def test():
    print("generator start")
    n = 1
    while True:
        yield_expression_value = yield n
        print("yield_expression_value = %d" % yield_expression_value)
        n += 1


# ①創(chuàng)建generator對(duì)象
generator = test()
print(type(generator))

print("\n---------------\n")

# ②啟動(dòng)generator
next_result = generator.__next__()
print("next_result = %d" % next_result)

print("\n---------------\n")

# ③發(fā)送值給yield表達(dá)式
send_result = generator.send(666)
print("send_result = %d" % send_result)

執(zhí)行結(jié)果：

<class 'generator'>

---------------

generator start
next_result = 1

---------------

yield_expression_value = 666
send_result = 2

方法說(shuō)明：

  • __next__()方法: 作用是啟動(dòng)或者恢復(fù)generator的執(zhí)行，相當(dāng)于send(None)

  • send(value)方法：作用是發(fā)送值給yield表達(dá)式。啟動(dòng)generator則是調(diào)用send(None)

執(zhí)行結(jié)果的說(shuō)明：

  • ①創(chuàng)建generator對(duì)象：包含yield表達(dá)式的函數(shù)將不再是一個(gè)函數(shù)，調(diào)用之后將會(huì)返回generator對(duì)象

  • ②啟動(dòng)generator：使用生成器之前需要先調(diào)用__next__或者send(None)，否則將報(bào)錯(cuò)。啟動(dòng)generator后，代碼將執(zhí)行到yield出現(xiàn)的位置，也就是執(zhí)行到yield n，然后將n傳遞到generator.__next__()這行的返回值。（注意，生成器執(zhí)行到yield n后將暫停在這里，直到下一次生成器被啟動(dòng)）

  • ③發(fā)送值給yield表達(dá)式：調(diào)用send方法可以發(fā)送值給yield表達(dá)式，同時(shí)恢復(fù)生成器的執(zhí)行。生成器從上次中斷的位置繼續(xù)向下執(zhí)行，然后遇到下一個(gè)yield，生成器再次暫停，切換到主函數(shù)打印出send_result。

理解這個(gè)demo的關(guān)鍵是：生成器啟動(dòng)或恢復(fù)執(zhí)行一次，將會(huì)在yield處暫停。上面的第②步僅僅執(zhí)行到了yield n，并沒(méi)有執(zhí)行到賦值語(yǔ)句，到了第③步，生成器恢復(fù)執(zhí)行才給yield_expression_value賦值。

生產(chǎn)者和消費(fèi)者模型

上面的例子中，代碼中斷-->切換執(zhí)行，體現(xiàn)出了協(xié)程的部分特點(diǎn)。

我們?cè)倥e一個(gè)生產(chǎn)者、消費(fèi)者的例子，這個(gè)例子來(lái)自廖雪峰的Python教程：

傳統(tǒng)的生產(chǎn)者-消費(fèi)者模型是一個(gè)線程寫(xiě)消息，一個(gè)線程取消息，通過(guò)鎖機(jī)制控制隊(duì)列和等待，但一不小心就可能死鎖。

現(xiàn)在改用協(xié)程，生產(chǎn)者生產(chǎn)消息后，直接通過(guò)yield跳轉(zhuǎn)到消費(fèi)者開(kāi)始執(zhí)行，待消費(fèi)者執(zhí)行完畢后，切換回生產(chǎn)者繼續(xù)生產(chǎn)，效率極高。

def consumer():
    print("[CONSUMER] start")
    r = 'start'
    while True:
        n = yield r
        if not n:
            print("n is empty")
            continue
        print("[CONSUMER] Consumer is consuming %s" % n)
        r = "200 ok"


def producer(c):
    # 啟動(dòng)generator
    start_value = c.send(None)
    print(start_value)
    n = 0
    while n < 3:
        n += 1
        print("[PRODUCER] Producer is producing %d" % n)
        r = c.send(n)
        print('[PRODUCER] Consumer return: %s' % r)
    # 關(guān)閉generator
    c.close()


# 創(chuàng)建生成器
c = consumer()
# 傳入generator
producer(c)

執(zhí)行結(jié)果：

[CONSUMER] start
start
[PRODUCER] producer is producing 1
[CONSUMER] consumer is consuming 1
[PRODUCER] Consumer return: 200 ok
[PRODUCER] producer is producing 2
[CONSUMER] consumer is consuming 2
[PRODUCER] Consumer return: 200 ok
[PRODUCER] producer is producing 3
[CONSUMER] consumer is consuming 3
[PRODUCER] Consumer return: 200 ok

注意到consumer函數(shù)是一個(gè)generator，把一個(gè)consumer傳入produce后：

  1. 首先調(diào)用c.send(None)啟動(dòng)生成器；
  2. 然后，一旦生產(chǎn)了東西，通過(guò)c.send(n)切換到consumer執(zhí)行；
  3. consumer通過(guò)yield拿到消息，處理，又通過(guò)yield把結(jié)果傳回；
  4. produce拿到consumer處理的結(jié)果，繼續(xù)生產(chǎn)下一條消息；
  5. produce決定不生產(chǎn)了，通過(guò)c.close()關(guān)閉consumer，整個(gè)過(guò)程結(jié)束。

整個(gè)流程無(wú)鎖，由一個(gè)線程執(zhí)行，produce和consumer協(xié)作完成任務(wù)，所以稱為“協(xié)程”，而非線程的搶占式多任務(wù)。

yield from表達(dá)式

Python3.3版本新增yield from語(yǔ)法，新語(yǔ)法用于將一個(gè)生成器部分操作委托給另一個(gè)生成器。此外，允許子生成器（即yield from后的“參數(shù)”）返回一個(gè)值，該值可供委派生成器（即包含yield from的生成器）使用。并且在委派生成器中，可對(duì)子生成器進(jìn)行優(yōu)化。

我們先來(lái)看最簡(jiǎn)單的應(yīng)用，例如：

# 子生成器
def test(n):
    i = 0
    while i < n:
        yield i
        i += 1

# 委派生成器
def test_yield_from(n):
    print("test_yield_from start")
    yield from test(n)
    print("test_yield_from end")


for i in test_yield_from(3):
    print(i)

輸出：

test_yield_from start
0
1
2
test_yield_from end

這里我們僅僅給這個(gè)生成器添加了一些打印，如果是正式的代碼中，你可以添加正常的執(zhí)行邏輯。

如果上面的test_yield_from函數(shù)中有兩個(gè)yield from語(yǔ)句，將串行執(zhí)行。比如將上面的test_yield_from函數(shù)改寫(xiě)成這樣：

def test_yield_from(n):
    print("test_yield_from start")
    yield from test(n)
    print("test_yield_from doing")
    yield from test(n)
    print("test_yield_from end")

將輸出：

test_yield_from start
0
1
2
test_yield_from doing
0
1
2
test_yield_from end

在這里，yield from起到的作用相當(dāng)于下面寫(xiě)法的簡(jiǎn)寫(xiě)形式

for item in test(n):
    yield item

看起來(lái)這個(gè)yield from也沒(méi)做什么大不了的事，其實(shí)它還幫我們處理了異常之類的。具體可以看stackoverflow上的這個(gè)問(wèn)題：In practice, what are the main uses for the new “yield from” syntax in Python 3.3?

協(xié)程(Coroutine)

  • Python3.4開(kāi)始，新增了asyncio相關(guān)的API，語(yǔ)法使用@asyncio.coroutine和yield from實(shí)現(xiàn)協(xié)程
  • Python3.5中引入async/await語(yǔ)法，參見(jiàn)PEP492

我們先來(lái)看Python3.4的實(shí)現(xiàn)。

@asyncio.coroutine

Python3.4中，使用@asyncio.coroutine裝飾的函數(shù)稱為協(xié)程。不過(guò)沒(méi)有從語(yǔ)法層面進(jìn)行嚴(yán)格約束。

對(duì)裝飾器不了解的小伙伴可以看我的上一篇博客--《理解Python裝飾器》

對(duì)于Python原生支持的協(xié)程來(lái)說(shuō)，Python對(duì)協(xié)程和生成器做了一些區(qū)分，便于消除這兩個(gè)不同但相關(guān)的概念的歧義：

  • 標(biāo)記了@asyncio.coroutine裝飾器的函數(shù)稱為協(xié)程函數(shù)，iscoroutinefunction()方法返回True
  • 調(diào)用協(xié)程函數(shù)返回的對(duì)象稱為協(xié)程對(duì)象，iscoroutine()函數(shù)返回True

舉個(gè)栗子，我們給上面yield from的demo中添加@asyncio.coroutine：

import asyncio

...

@asyncio.coroutine
def test_yield_from(n):
    ...

# 是否是協(xié)程函數(shù)
print(asyncio.iscoroutinefunction(test_yield_from))
# 是否是協(xié)程對(duì)象
print(asyncio.iscoroutine(test_yield_from(3)))

毫無(wú)疑問(wèn)輸出結(jié)果是True。

可以看下@asyncio.coroutine的源碼中查看其做了什么，我將其源碼簡(jiǎn)化下，大致如下：

import functools
import types
import inspect

def coroutine(func):
    # 判斷是否是生成器
    if inspect.isgeneratorfunction(func):
        coro = func
    else:
        # 將普通函數(shù)變成generator
        @functools.wraps(func)
        def coro(*args, **kw):
            res = func(*args, **kw)
            res = yield from res
            return res
    # 將generator轉(zhuǎn)換成coroutine
    wrapper = types.coroutine(coro)
    # For iscoroutinefunction().
    wrapper._is_coroutine = True
    return wrapper

將這個(gè)裝飾器標(biāo)記在一個(gè)生成器上，就會(huì)將其轉(zhuǎn)換成coroutine。

然后，我們來(lái)實(shí)際使用下@asyncio.coroutine和yield from：

import asyncio

@asyncio.coroutine
def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    yield from asyncio.sleep(1.0)
    return x + y

@asyncio.coroutine
def print_sum(x, y):
    result = yield from compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
print("start")
# 中斷調(diào)用，直到協(xié)程執(zhí)行結(jié)束
loop.run_until_complete(print_sum(1, 2))
print("end")
loop.close()

執(zhí)行結(jié)果：

start
Compute 1 + 2 ...
1 + 2 = 3
end

print_sum這個(gè)協(xié)程中調(diào)用了子協(xié)程compute，它將等待compute執(zhí)行結(jié)束才返回結(jié)果。

這個(gè)demo點(diǎn)調(diào)用流程如下圖：

tulip_coro.png

EventLoop將會(huì)把print_sum封裝成Task對(duì)象

流程圖展示了這個(gè)demo的控制流程，不過(guò)沒(méi)有展示其全部細(xì)節(jié)。比如其中“暫?！钡?s，實(shí)際上創(chuàng)建了一個(gè)future對(duì)象, 然后通過(guò)BaseEventLoop.call_later()在1s后喚醒這個(gè)任務(wù)。

值得注意的是，@asyncio.coroutine將在Python3.10版本中移除。

async/await

Python3.5開(kāi)始引入async/await語(yǔ)法（PEP 492），用來(lái)簡(jiǎn)化協(xié)程的使用并且便于理解。

async/await實(shí)際上只是@asyncio.coroutine和yield from的語(yǔ)法糖：

  • 把@asyncio.coroutine替換為async
  • 把yield from替換為await

即可。

比如上面的例子：

import asyncio


async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y


async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))


loop = asyncio.get_event_loop()
print("start")
loop.run_until_complete(print_sum(1, 2))
print("end")
loop.close()

我們?cè)賮?lái)看一個(gè)asyncio中Future的例子：

import asyncio

future = asyncio.Future()


async def coro1():
    print("wait 1 second")
    await asyncio.sleep(1)
    print("set_result")
    future.set_result('data')


async def coro2():
    result = await future
    print(result)


loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait([
    coro1()
    coro2()
]))
loop.close()

輸出結(jié)果：

wait 1 second
(大約等待1秒)
set_result
data

這里await后面跟隨的future對(duì)象，協(xié)程中yield from或者await后面可以調(diào)用future對(duì)象，其作用是：暫停協(xié)程，直到future執(zhí)行結(jié)束或者返回result或拋出異常。

而在我們的例子中，await future必須要等待future.set_result('data')后才能夠結(jié)束。將coro2()作為第二個(gè)協(xié)程可能體現(xiàn)得不夠明顯，可以將協(xié)程的調(diào)用改成這樣：

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait([
    # coro1(),
    coro2(),
    coro1()
]))
loop.close()

輸出的結(jié)果仍舊與上面相同。

其實(shí)，async這個(gè)關(guān)鍵字的用法不止能用在函數(shù)上，還有async with異步上下文管理器，async for異步迭代器. 對(duì)這些感興趣且覺(jué)得有用的可以網(wǎng)上找找資料，這里限于篇幅就不過(guò)多展開(kāi)了。

總結(jié)

本文就生成器和協(xié)程做了一些學(xué)習(xí)、探究和總結(jié)，不過(guò)并沒(méi)有做過(guò)多深入的研究。權(quán)且作為入門(mén)到一個(gè)筆記，之后將會(huì)嘗試自己實(shí)現(xiàn)一下異步API，希望有助于理解學(xué)習(xí)。

參考鏈接

Python協(xié)程 https://thief.one/2017/02/20/Python%E5%8D%8F%E7%A8%8B/

http://www.dabeaz.com/coroutines/Coroutines.pdf

Coroutines

How the heck does async/await work in Python 3.5

Python3.4協(xié)程文檔

Python3.5協(xié)程文檔

廖雪峰的Python教程--協(xié)程

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容