7.11 內(nèi)聯(lián)回調(diào)函數(shù)

問(wèn)題

當(dāng)你編寫(xiě)使用回調(diào)函數(shù)的代碼的時(shí)候，擔(dān)心很多小函數(shù)的擴(kuò)張可能會(huì)弄亂程序控制流。你希望找到某個(gè)方法來(lái)讓代碼看上去更像是一個(gè)普通的執(zhí)行序列。

解決方案

通過(guò)使用生成器和協(xié)程可以使得回調(diào)函數(shù)內(nèi)聯(lián)在某個(gè)函數(shù)中。為了演示說(shuō)明，假設(shè)你有如下所示的一個(gè)執(zhí)行某種計(jì)算任務(wù)然后調(diào)用一個(gè)回調(diào)函數(shù)的函數(shù)(參考7.10小節(jié))：

def apply_async(func, args, *, callback):
    # Compute the result
    result = func(*args)

    # Invoke the callback with the result
    callback(result)

接下來(lái)讓我們看一下下面的代碼，它包含了一個(gè) Async 類和一個(gè) inlined_async 裝飾器：

from queue import Queue
from functools import wraps

class Async:
    def __init__(self, func, args):
        self.func = func
        self.args = args

def inlined_async(func):
    @wraps(func)
    def wrapper(*args):
        f = func(*args)
        result_queue = Queue()
        result_queue.put(None)
        while True:
            result = result_queue.get()
            try:
                a = f.send(result)
                apply_async(a.func, a.args, callback=result_queue.put)
            except StopIteration:
                break
    return wrapper

這兩個(gè)代碼片段允許你使用 yield 語(yǔ)句內(nèi)聯(lián)回調(diào)步驟。比如：

def add(x, y):
    return x + y

@inlined_async
def test():
    r = yield Async(add, (2, 3))
    print(r)
    r = yield Async(add, ('hello', 'world'))
    print(r)
    for n in range(10):
        r = yield Async(add, (n, n))
        print(r)
    print('Goodbye')

如果你調(diào)用 test() ，你會(huì)得到類似如下的輸出：

5
helloworld
0
2
4
6
8
10
12
14
16
18
Goodbye

你會(huì)發(fā)現(xiàn)，除了那個(gè)特別的裝飾器和 yield 語(yǔ)句外，其他地方并沒(méi)有出現(xiàn)任何的回調(diào)函數(shù)(其實(shí)是在后臺(tái)定義的)。

討論

本小節(jié)會(huì)實(shí)實(shí)在在的測(cè)試你關(guān)于回調(diào)函數(shù)、生成器和控制流的知識(shí)。

首先，在需要使用到回調(diào)的代碼中，關(guān)鍵點(diǎn)在于當(dāng)前計(jì)算工作會(huì)掛起并在將來(lái)的某個(gè)時(shí)候重啟(比如異步執(zhí)行)。當(dāng)計(jì)算重啟時(shí)，回調(diào)函數(shù)被調(diào)用來(lái)繼續(xù)處理結(jié)果。apply_async() 函數(shù)演示了執(zhí)行回調(diào)的實(shí)際邏輯，盡管實(shí)際情況中它可能會(huì)更加復(fù)雜(包括線程、進(jìn)程、事件處理器等等)。

計(jì)算的暫停與重啟思路跟生成器函數(shù)的執(zhí)行模型不謀而合。具體來(lái)講，yield 操作會(huì)使一個(gè)生成器函數(shù)產(chǎn)生一個(gè)值并暫停。接下來(lái)調(diào)用生成器的 __next__() 或 send() 方法又會(huì)讓它從暫停處繼續(xù)執(zhí)行。

根據(jù)這個(gè)思路，這一小節(jié)的核心就在 inline_async() 裝飾器函數(shù)中了。關(guān)鍵點(diǎn)就是，裝飾器會(huì)逐步遍歷生成器函數(shù)的所有 yield 語(yǔ)句，每一次一個(gè)。為了這樣做，剛開(kāi)始的時(shí)候創(chuàng)建了一個(gè) result 隊(duì)列并向里面放入一個(gè) None 值。然后開(kāi)始一個(gè)循環(huán)操作，從隊(duì)列中取出結(jié)果值并發(fā)送給生成器，它會(huì)持續(xù)到下一個(gè) yield 語(yǔ)句，在這里一個(gè) Async 的實(shí)例被接受到。然后循環(huán)開(kāi)始檢查函數(shù)和參數(shù)，并開(kāi)始進(jìn)行異步計(jì)算 apply_async() 。然而，這個(gè)計(jì)算有個(gè)最詭異部分是它并沒(méi)有使用一個(gè)普通的回調(diào)函數(shù)，而是用隊(duì)列的 put() 方法來(lái)回調(diào)。

這時(shí)候，是時(shí)候詳細(xì)解釋下到底發(fā)生了什么了。主循環(huán)立即返回頂部并在隊(duì)列上執(zhí)行 get() 操作。如果數(shù)據(jù)存在，它一定是 put() 回調(diào)存放的結(jié)果。如果沒(méi)有數(shù)據(jù)，那么先暫停操作并等待結(jié)果的到來(lái)。這個(gè)具體怎樣實(shí)現(xiàn)是由 apply_async() 函數(shù)來(lái)決定的。如果你不相信會(huì)有這么神奇的事情，你可以使用 multiprocessing 庫(kù)來(lái)試一下，在單獨(dú)的進(jìn)程中執(zhí)行異步計(jì)算操作，如下所示：

if __name__ == '__main__':
    import multiprocessing
    pool = multiprocessing.Pool()
    apply_async = pool.apply_async

    # Run the test function
    test()

實(shí)際上你會(huì)發(fā)現(xiàn)這個(gè)真的就是這樣的，但是要解釋清楚具體的控制流得需要點(diǎn)時(shí)間了。

將復(fù)雜的控制流隱藏到生成器函數(shù)背后的例子在標(biāo)準(zhǔn)庫(kù)和第三方包中都能看到。比如，在contextlib 中的 @contextmanager 裝飾器使用了一個(gè)令人費(fèi)解的技巧，通過(guò)一個(gè) yield 語(yǔ)句將進(jìn)入和離開(kāi)上下文管理器粘合在一起。另外非常流行的 Twisted 包中也包含了非常類似的內(nèi)聯(lián)回調(diào)。