（轉） Twisted ：第十八部分 Deferreds 全貌

簡介

在上一個部分,咱們學習了使用生成器構造順序異步回調的新方法.這樣,包括 deferreds,咱們如今有兩種將異步操做連接在一塊兒的方法.

有時,然而,咱們須要"並行"的運行一組異步操做.因爲Twisted是單線程的,它實際並不會併發運行,但咱們但願使用異步I/O在一組任務上儘量快的工做.以咱們的詩歌客戶端爲例,它從多個服務器同時下載詩歌,而不是一個接一個的方式.這就是使用Twisted下載詩歌的所有特色.

做爲一個結果,全部詩歌客戶端須要解決一個問題:你怎樣得知你啓動的全部異步操做已經完成?目前咱們經過將結果集總到一個列表(如客戶端 7.0中的 結果 列表)並檢查這個列表的長度來解決這個問題.除了收集成功的結果,咱們還必須當心地對待失敗,不然一個失敗將使程序進入死循環,覺得還有工做須要作.

正如你所料,Twisted包含一個抽象層能夠用來解決這個問題,咱們來看一看.

DeferredList

DeferredList 類使咱們能夠將一個 defered 對象列表視爲一個 defered 對象.經過這種方法咱們啓動一族異步操做而且在它們所有完成後得到通知(不管它們成功或者失敗).讓咱們看一些例子.

在 deferred-list/deferred-list-1.py 中,能夠找到以下代碼:

from twisted.internet import defer

def got_results(res):
    print 'We got:', res

print 'Empty List.'
d = defer.DeferredList([])
print 'Adding Callback.'
d.addCallback(got_results)

若是運行它,將獲得以下輸出:

Empty List.
Adding Callback.
We got: []

注意如下幾點:

• DeferredList 由Python列表建立.在這種狀況下,列表是空的,但咱們很快將看到列表元素必須是 Deferred 對象.

• DeferredList 自己是一個 deferred (它繼承 Deferred).這意味着你能夠像對待普通 deferred 同樣向其添加回調和錯誤回調.

• 在以上例子中,回調被添加時當即激發,因此 DeferredList 也必須當即激發.咱們一下子將討論.

• deferred 列表的結果自己也是一個列表(空).

下面看一下 deferred-list/deferred-list-2.py:

from twisted.internet import defer

def got_results(res):
    print 'We got:', res

print 'One Deferred.'
d1 = defer.Deferred()
d = defer.DeferredList([d1])
print 'Adding Callback.'
d.addCallback(got_results)
print 'Firing d1.'
d1.callback('d1 result')

如今咱們建立了包含一個 deferred 元素的 DeferredList 列表,獲得以下輸出:

One Deferred.
Adding Callback.
Firing d1.
We got: [(True, 'd1 result')]

注意如下幾點:

• 此次 DeferredList 沒有激發它的回調,直到咱們激發列表中的 deferred.

• 結果一樣是一個列表,但此次包含一個元素.

• 這個元素是一個元組,它的第二個值是列表中 deferred 的結果.

讓咱們向列表添加兩個 deferreds (deferred-list/deferred-list-3.py):

from twisted.internet import defer

def got_results(res):
    print 'We got:', res

print 'Two Deferreds.'
d1 = defer.Deferred()
d2 = defer.Deferred()
d = defer.DeferredList([d1, d2])
print 'Adding Callback.'
d.addCallback(got_results)
print 'Firing d1.'
d1.callback('d1 result')
print 'Firing d2.'
d2.callback('d2 result')

獲得以下輸出:

Two Deferreds.
Adding Callback.
Firing d1.
Firing d2.
We got: [(True, 'd1 result'), (True, 'd2 result')]

如今 DeferredList 的結果很是清晰,至少以咱們的使用方式,它是一個列表,元素個數與傳入構造器的 deferred 列表元素個數相同. 並且結果列表的元素包含原始的 deferreds 結果信息,至少當這些 deferred 成功返回.這意味着 DeferredList 自己並不激發直到全部的原始列表中的 deferreds 都被激發. 並且以一個空列表建立的 DeferredList 會當即激發,由於它不須要等待任何 deferreds.

那麼最終結果列表中的元素順序如何? 考慮如下代碼( deferred-list/deferred-list-4.py):

from twisted.internet import defer

def got_results(res):
    print 'We got:', res

print 'Two Deferreds.'
d1 = defer.Deferred()
d2 = defer.Deferred()
d = defer.DeferredList([d1, d2])
print 'Adding Callback.'
d.addCallback(got_results)
print 'Firing d2.'
d2.callback('d2 result')
print 'Firing d1.'
d1.callback('d1 result')

這裏咱們先激發 d2 而後再激發 d1,注意構造參數中的 deferred 列表裏 d1, d2 還是原先的順序.輸出結果以下:

Two Deferreds.
Adding Callback.
Firing d2.
Firing d1.
We got: [(True, 'd1 result'), (True, 'd2 result')]

輸出列表中結果的順序與原始 deferred 列表順序相對應,而不是 deferred 碰巧被激發的順序.這一點很是好,由於咱們能夠很容易地將每一個結果與生成它的相應的操做聯繫在一塊兒(如哪首詩來自哪一個服務器).

好了,那若是列表中一個或多個 deferreds 失敗了怎麼辦呢? 上面結果中的 True 有什麼用? 再看一個例子(deferred-list/deferred-list-5.py):

from twisted.internet import defer

def got_results(res):
    print 'We got:', res

d1 = defer.Deferred()
d2 = defer.Deferred()
d = defer.DeferredList([d1, d2], consumeErrors=True)
d.addCallback(got_results)
print 'Firing d1.'
d1.callback('d1 result')
print 'Firing d2 with errback.'
d2.errback(Exception('d2 failure'))

如今咱們以正常結果激發 d1,以錯誤激發 d2.先暫時忽略 consumerErrors 選項,稍候介紹.這裏是輸出結果:

Firing d1.
Firing d2 with errback.
We got: [(True, 'd1 result'), (False, <twisted.python.failure.Failure <type 'exceptions.Exception'>>)]

此次對應 d2 的元組在第二個位置出現了一個 Failure,而且第一個位置是 False.至此 DeferredList 的工做原理很是清晰(但繼續瀏覽如下討論):

• DeferredList 是以一個 deferred 對象列表建立的.

• DeferredList 自己是一個 deferred,它返回的結果是一個列表,長度與 deferred 列表相同.

• 當原始列表中全部 deferred 被激發後, DeferredList 將會被激發.

• 結果列表中的每一個元素以相同順序對應原始列表中相應的 deferred.若是那個 deferred 成功返回,相應元素是(True,result),若是失敗則爲(False,failure).

• DeferredList 不會失敗,由於不管每一個 deferred 的返回結果是什麼都會被集總到結果列表中(一樣,請看下面討論).

如今讓咱們討論一下被傳入 DeferredList 的 consumeErrors 選項,若是咱們運行以上相同代碼而不傳入此選項(deferred-list/deferred-list-6.py),則獲得如下輸出:

Firing d1.
Firing d2 with errback.
We got: [(True, 'd1 result'), (False, >twisted.python.failure.Failure >type 'exceptions.Exception'<<)]
Unhandled error in Deferred:
Traceback (most recent call last):
Failure: exceptions.Exception: d2 failure

若是你還記得,"Unhandled error in Deferred"消息是在 deferred 垃圾回收時被生成的,並且它表示最後一個回調失敗了.這個消息告訴咱們並無徹底捕獲潛在的異步錯誤.在咱們例子中,它是從哪裏來的呢? 很明顯不是來自 DeferredList,由於它已經成功返回了.因此它必定是來自 d2.

DeferredList 須要知道它所監視的 deferred 什麼時候激發. DeferredList 以一般的方式向每一個 deferred 添加一個回調和錯誤回調. 默認地,這個回調(或錯誤)返回原始結果(或錯誤)在將它們放入最終結果列表以後.因爲錯誤回調返回原始 failure 後將觸發下一個錯誤回調, d2 在它被激發後仍然保持失敗狀態.

可是若是咱們將 consumeErrors=True 傳遞給 DeferredList, 它將向每一個 deferred 添加返回 None 的錯誤回調, 即"消耗"掉這個錯誤而且取消警告信息. 咱們一樣能夠向 d2 添加本身的錯誤回調來處理錯誤,如 deferred-list/deferred-list-7.py.

客戶端 8.0

獲取詩歌客戶端8.0發佈啦！客戶端使用 DeferredList 去發現全部詩歌什麼時候完成(或失敗).新版客戶端位於 twisted-client-8/get-poetry.py. 一樣,惟一的變化在於 poetry_main, 咱們來看一下重要的變化:

...
ds = []

for (host, port) in addresses:
    d = get_transformed_poem(host, port)
    d.addCallbacks(got_poem)
    ds.append(d)

dlist = defer.DeferredList(ds, consumeErrors=True)
dlist.addCallback(lambda res : reactor.stop())

你能夠與 客戶端 7.0 中的相應部分比較.

在客戶端 8.0中,咱們不須要 poem_done 回調和 results 列表.相反,咱們把每一個從 get_transformed_poem 返回的 deferred 放入 ds 列表,以後建立一個 DeferredList.因爲 DeferredList 不會在全部詩歌完成或失敗以前激發,咱們僅僅向 DeferredList 添加一個回調以便關閉 reactor. 在咱們這個狀況中,沒有使用 DeferredList 返回的結果,咱們僅僅須要知道全部事情什麼時候結束.僅此而已!

討論

可視化 DeferredList 的工做方式:

圖37: DeferredList 的結果

很是簡單,真的. 還有一些關於 DeferredList 的選項咱們沒有涉及,以及那些改變咱們以上所描述行爲的選項.咱們在參考練習中把這些留給讀者本身探索.

在 :doc:`p19` 中咱們將進一步介紹 Deferred 類, 包括 Twisted 10.1.0 提出的最新特性.

參考練習

1. 閱讀 DeferredList 的源代碼.

2. 修改 deferred-list 中的例子去實現可選的構造器參數 fireOnOneCallback 和 fireOnOneErrback. 實現你將用其中一個(或兩個都使用)的情景.

3. 你可使用 DeferredLists 列表建立一個 DeferredList 嗎? 若是是這樣,結果將是什麼?

4. 修改客戶端8.0在全部詩歌完成下載前不打印任意信息. 此次你將使用 DeferredList 的結果.

5. 定義 DeferredDict 的句法而且實現它.