Twisted是一個正在進展的項目,它的開發(fā)者會定期添加新的特性并且擴展舊的特性.
隨著Twisted 10.1.0發(fā)布,開發(fā)者向 Deferred 類添加了一個新的特性—— cancellation ——這正是我們今天要研究的.
異步編程將請求和響應解耦了,如此又帶來一個新的可能性:在請求結果和返回結果之間,你可能決定不再需要這個結果了.考慮一下 第十四節(jié) 中的詩歌代理服務器.下面展示代理如何工作的,至少對于詩歌的第一次請求:
看起來非常完美,但是如果客戶端在獲得詩歌之前斷開了鏈接怎么辦?也許它們先前請求 Paradise Lost的全部內(nèi)容,隨后它們決定實際想要的是 Kojo的俳句.我們的代理還在傻傻的處理前面的下載請求,而且那個緩慢的下載服務器還要等好一會.最好的策略便是關閉連接,讓下載服務器回去順覺.
回憶一下 圖15 展示了同步程序控制流的概念.在那張圖中我們可以看到函數(shù)調(diào)用自上而下,異常也是自下而上.如果我們希望取消一個同步調(diào)用(這僅是假設),控制流的傳遞方向與函數(shù)調(diào)用的方向一致,都是從高層傳向底層,如圖38所示:
http://wiki.jikexueyuan.com/project/twisted-intro/images/p19_sync-cancel.png" alt="" /> 圖38 同步程序流,含假想取消操作
當然,在同步程序中這是不可能的,因為高層的代碼在底層操作結束前沒有恢復運行,自然也就沒有什么可取消的.但是在異步程序中,高層代碼在底層代碼完成前具有控制權,至少具有在底層代碼完成之前取消它的請求的可能性.
在Twisted程序中,底層請求被包含在一個 Deferred 對象中,你可以將其想象為一個外部異步操作的"句柄". deferred 中正常的信息流是向下的,從底層代碼到高層代碼,與同步程序中返回的信息流方向一致.從Twisted 10.1.0開始,高層代碼可以反向發(fā)送信息 —— 它可以告訴底層代碼它不再需要其結果了.如圖39:
http://wiki.jikexueyuan.com/project/twisted-intro/images/p19_deferred-cancel.png" alt="" /> 圖39 deferred 中的信息流,包含取消
讓我們看一些例程,來了解下取消 deferreds 的實際工作原理.注意為了運行這些列子以及本部分中的其他代碼,你需要安裝Twisted 10.1.0或更高 版本 考慮 deferred-cancel/defer-cancel-1.py
from twisted.internet import defer
def callback(res):
print 'callback got:', res
d = defer.Deferred()
d.addCallback(callback)
d.cancel()
print 'done'伴隨著新的取消特性, Deferred 類添加了一個名為 cancel 的新方法.上面代碼創(chuàng)建了一個新的 deferred,添加了一個回調(diào),然后取消了這個 deferred 而沒有激發(fā)它.輸出如下:
done
Unhandled error in Deferred:
Traceback (most recent call last):
Failure: twisted.internet.defer.CancelledError:OK,取消一個 deferred 看起來像使錯誤回調(diào)鏈運行,常規(guī)的回調(diào)根本沒有被調(diào)用.同樣注意到這個錯誤是: twisted.internet.defer.CancelledError,一個意味著 deferred 被取消的個性化異常(但請繼續(xù)閱讀).讓我們添加一個錯誤回調(diào),如 [deferred-cancel/defer-cancel-2.py](https://github.com/jdavisp3/twisted-intro/blob/master/deferred-cancel/defer-cancel-2.py#L1>`_
from twisted.internet import defer
def callback(res):
print 'callback got:', res
def errback(err):
print 'errback got:', err
d = defer.Deferred()
d.addCallbacks(callback, errback)
d.cancel()
print 'done'得到以下輸出:
errback got: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]
done所以我們可以'捕獲'從 cancel 產(chǎn)生的錯誤回調(diào),就像其他 deferred 錯誤一樣.
OK,讓我們試試激發(fā) deferred 然后取消它,如 deferred-cancel/defer-cancel-3.py
from twisted.internet import defer
def callback(res):
print 'callback got:', res
def errback(err):
print 'errback got:', err
d = defer.Deferred()
d.addCallbacks(callback, errback)
d.callback('result')
d.cancel()
print 'done'這里我們用常規(guī) callback 方法激發(fā) deferred,之后取消它.輸出結果如下:
callback got: result
done我們的回調(diào)被調(diào)用(正如我們所預期的)之后程序正常結束,就像 cancel 根本沒有被調(diào)用.所以取消一個已經(jīng)被激發(fā)的 deferred 沒有任何效果(但請繼續(xù)閱讀!).
如果我們在取消 deferred 之后激發(fā)它會怎樣?參看 deferred-cancel/defer-cancel-4.py
from twisted.internet import defer
def callback(res):
print 'callback got:', res
def errback(err):
print 'errback got:', err
d = defer.Deferred()
d.addCallbacks(callback, errback)
d.cancel()
d.callback('result')
print 'done'這種情況的輸出如下:
errback got: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]
done有意思！與第二個例子的輸出一樣,當時沒有激發(fā) deferred.所以如果 deferred 被取消了,再激發(fā)它沒有效果.但是為什么 d.callback('result') 沒有產(chǎn)生錯誤,考慮到不能激發(fā) deferred 大于一次,錯誤回調(diào)鏈為何沒有運行?
再次考慮 圖39，用結果或失敗激發(fā)一個 deferred 是底層代碼的工作,然而取消 deferred 是高層代碼的行為.激發(fā) deferred 意味著"這是你的結果",然而取消 deferred 意味著"我不再想要這個結果了".同時記住 cancel 是一個新特性,所以大部分現(xiàn)有的Twisted代碼并沒有處理取消的操作.但是Twisted的開發(fā)者使我們?nèi)∠?deferred 的想法變得有可能,甚至包括那些在Twisted 10.1.0之前寫的代碼.
為了實現(xiàn)以上想法, cancel 方法實際上做兩件事:
Deferred 對象你不想要那個結果,如果它還沒有返回(如, deferred 沒有被激發(fā)),忽略后續(xù)任何回調(diào)或錯誤回調(diào)的調(diào)用.由于舊版本的Twisted代碼會上前去激發(fā)任何已經(jīng)被取消的 deferred, 步驟1 確保我們的程序不會垮掉如果我們?nèi)∠粋€舊有庫中的 deferred.
這意味著我們可以隨心所欲地取消一個 deferred,同時可以確定不會得到結果如果它還沒有到來(甚至那些 將要 到來的).但是取消 deferred 可能并沒有取消異步操作.終止一個異步操作需要一個上下文的具體行動.你可能需要關閉網(wǎng)絡連接,回滾數(shù)據(jù)庫事務,結束子進程,等等.由于 deferred 僅僅是一般目的的回調(diào)組織者,它怎么知道具體要做什么當你取消它時?或者,換種說法,它怎樣將 cancel 請求傳遞給首先已經(jīng)創(chuàng)建和返回了 deferred 的底層代碼? 和我一起說:
I know, with a callback!好吧,首先看一下 deferred-cancel/defer-cancel-5.py
from twisted.internet import defer
def canceller(d):
print "I need to cancel this deferred:", d
def callback(res):
print 'callback got:', res
def errback(err):
print 'errback got:', err
d = defer.Deferred(canceller) # created by lower-level code
d.addCallbacks(callback, errback) # added by higher-level code
d.cancel()
print 'done'這個例子基本上跟第二個例子相同,除了有第三個回調(diào)(canceller).這個回調(diào)是我們在創(chuàng)建 Deferred 的時候傳遞給它的,不是之后添加的.這個回調(diào)負責執(zhí)行終止異步操作時所需的上下文相關的具體操作(當然,僅當 deferred 被實際取消). canceller 回調(diào)是返回 deferred 的底層代碼的必要部分,不是接收 deferred 的高層代碼為其自己添加的回調(diào)和錯誤回調(diào).
運行這個例子將產(chǎn)生如下輸出:
I need to cancel this deferred: <Deferred at 0xb7669d2cL>
errback got: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]
done正如你所看到, 不需要返回結果的 deferred 被傳遞給 canceller 回調(diào).在這里我們可以做任何需要做的事情以便徹底終止異步操作.注意 canceller 在錯誤回調(diào)鏈激發(fā)前被調(diào)用.其實我們可以在取消回調(diào)中選擇使用任何結果或錯誤自己激發(fā) deferred (這樣就會優(yōu)先于 CancelledError 失敗).這兩種情況在 deferred-cancel/defer-cancel-6.py 和 deferred-cancel/defer-cancel-7.py中進行了說明.
在激發(fā) reactor 之前先做一個簡單的測試.我們將使用 canceller 回調(diào)創(chuàng)建一個 deferred,正常的激發(fā)它,之后取消它.你可以在 deferred-cancel/defer-cancel-8.py中看到代碼.通過檢查那個腳本的輸出,你將看到取消一個被激發(fā)的 deferred 不會調(diào)用 canceller 回調(diào).這正是我們所要的,因為沒什么可取消的.
我們目前看到的例子都沒有實際的異步操作. 讓我們構造一個調(diào)用異步操作的簡單程序,之后我們將指出如何使那個操作可取消.參見代碼 deferred-cancel/defer-cancel-9.py
from twisted.internet.defer import Deferred
def send_poem(d):
print 'Sending poem'
d.callback('Once upon a midnight dreary')
def get_poem():
"""Return a poem 5 seconds later."""
from twisted.internet import reactor
d = Deferred()
reactor.callLater(5, send_poem, d)
return d
def got_poem(poem):
print 'I got a poem:', poem
def poem_error(err):
print 'get_poem failed:', err
def main():
from twisted.internet import reactor
reactor.callLater(10, reactor.stop) # stop the reactor in 10 seconds
d = get_poem()
d.addCallbacks(got_poem, poem_error)
reactor.run()
main()這個例子中包含了一個 get_poem 函數(shù),它使用 reactor 的 callLater 方法在被調(diào)用5秒鐘后異步地返回一首詩.主函數(shù)調(diào)用 get_poem,添加一個回調(diào)/錯誤回調(diào)對,之后啟動 reactor.我們(同樣使用 callLater)安排 reactor 在10秒鐘之后停止.通常我們向 deferred 添加一個回調(diào)來實現(xiàn),但你很快就會知道我們?yōu)楹芜@樣做.
運行程序(適當延遲后)產(chǎn)生如下輸出:
Sending poem
I got a poem: Once upon a midnight dreary10秒鐘后程序終止.現(xiàn)在來試試在詩歌被發(fā)送前取消 deferred.只需加入以下代碼在2秒鐘后取消(在5秒鐘延遲發(fā)送詩歌之前):
reactor.callLater(2, d.cancel) # cancel after 2 seconds 完整的例子參見 deferred-cancel/defer-cancel-10.py這將產(chǎn)生如下輸出:
get_poem failed: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]
Sending poem這個例子清晰地展示了取消一個 deferred 并沒有取消它背后的異步請求.2秒鐘后我們看到了錯誤回調(diào)輸出,打印出如我們所料的 CancelledError 錯誤.但是5秒鐘后我們看到了 send_poem 的輸出(但是這個 deferred 上的回調(diào)并沒有激發(fā)).
這時我們與 deferred-cancel/defer-cancel-4.py的情況一樣."取消" deferred 僅僅是使最終結果被忽略,但實際上并沒有終止這個操作.正如我們上面所學,為了得到一個真正可取消的 deferred,必須在它被創(chuàng)建時添加一個 cancel 回調(diào).
那么這個新的回調(diào)需要做什么呢? 參考一下關于 callLater 方法的 文檔 它的返回值是另一個實現(xiàn)了 IDelayedCall 的對象,用 cancel 方法我們可以阻止延遲的調(diào)用被執(zhí)行.
這非常簡單,更新后的代碼參見 deferred-cancel/defer-cancel-11.py所有相關變化都在 get_poem 函數(shù)中:
def get_poem():
"""Return a poem 5 seconds later."""
def canceler(d):
# They don't want the poem anymore, so cancel the delayed call
delayed_call.cancel()
# At this point we have three choices:
# 1. Do nothing, and the deferred will fire the errback
# chain with CancelledError.
# 2. Fire the errback chain with a different error.
# 3. Fire the callback chain with an alternative result.
d = Deferred(canceler)
from twisted.internet import reactor
delayed_call = reactor.callLater(5, send_poem, d)
return d在這個新版本中,我們保存 callLater 的返回值以便能夠在 cancel 回調(diào)中使用. cancel 回調(diào)的唯一工作是調(diào)用 delayed_call.cancel(). 但是正如之前討論的,我們可以選擇激發(fā)自定義的 deferred. 最新版本的程序產(chǎn)生如下輸出:
get_poem failed: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]正如你看到的, deferred 被取消了并且異步操作被真正地終止了(我們看不到 send_poem 的輸出了).
正如在簡介中所討論,詩歌代理服務器是實現(xiàn)取消的很好的候選者,因為這可以讓我們?nèi)∠姼柘螺d如果事實證明沒有人想要它(如客戶端已經(jīng)在我們發(fā)送詩歌前關閉了連接).版本 3.0的代理位于 twisted-server-4/poetry-proxy.py實現(xiàn)了 deferred 取消. 變化首先位于 PoetryProxyProtocol
class PoetryProxyProtocol(Protocol):
def connectionMade(self):
self.deferred = self.factory.service.get_poem()
self.deferred.addCallback(self.transport.write)
self.deferred.addBoth(lambda r: self.transport.loseConnection())
def connectionLost(self, reason):
if self.deferred is not None:
deferred, self.deferred = self.deferred, None
deferred.cancel() # cancel the deferred if it hasn't fired你可以與 舊版本對比一下.兩個主要的變化是:
get_poem 得到的 deferred,以便之后在需要時取消它.deferred.注意當我們實際得到詩歌之后(鏈接最終肯定要關閉)，這個操作同樣會取消 deferred,但正如前例所發(fā)現(xiàn)的,取消一個被激發(fā)的 deferred 不會有任何效果. 現(xiàn)在我們需要確保取消 deferred 將實際終止詩歌的下載. 所以我們需要改變 ProxyService
class ProxyService(object):
poem = None # the cached poem
def __init__(self, host, port):
self.host = host
self.port = port
def get_poem(self):
if self.poem is not None:
print 'Using cached poem.'
# return an already-fired deferred
return succeed(self.poem)
def canceler(d):
print 'Canceling poem download.'
factory.deferred = None
connector.disconnect()
print 'Fetching poem from server.'
deferred = Deferred(canceler)
deferred.addCallback(self.set_poem)
factory = PoetryClientFactory(deferred)
from twisted.internet import reactor
connector = reactor.connectTCP(self.host, self.port, factory)
return factory.deferred
def set_poem(self, poem):
self.poem = poem
return poem同樣,可以與 舊版本對比一下. 這個類具有一些新的變化:
reactor.connetTCP 的返回值,一個 IConnector對象.我們可以使用這個對象上的 disconnect 方法關閉連接.canceler 回調(diào)的 deferred.那個回調(diào)是一個閉包,它使用 connector 關閉連接. 但首先須設置 factory.deferred 屬性為 None. 否則,工廠會以 "連接關閉"錯誤回調(diào)激發(fā) deferred 而不是以 CancelledError 激發(fā). 由于 deferred 已經(jīng)被取消了, 以 CancelledError 激發(fā)更加合適.你同樣會注意到我們是在 ProxyService 中創(chuàng)建 deferred 而不是 PoetryClientFactory. 由于 canceler 回調(diào)需要獲取 IConnector 對象, ProxyService 成為最方便創(chuàng)建 deferred 的地方.
同時,就像我們之前的例子, canceler 回調(diào)作為一個閉包實現(xiàn).閉包看起來在取消回調(diào)的實現(xiàn)上非常有用.
讓我們試試新的代理.首先啟動一個慢速服務器.它需要很慢以便我們有時間取消:
python blocking-server/slowpoetry.py --port 10001 poetry/fascination.txt現(xiàn)在可以啟動代理(記住你需要Twisted 10.1.0):
python twisted-server-4/poetry-proxy.py --port 10000 10001現(xiàn)在我們可以用任何客戶端從代理下載一首詩,或者僅使用 curl:
curl localhost:10000幾秒鐘后,按 Ctrl-C 停止客戶端或者 curl 進程. 在終端運行代理你將看到如下輸出:
Fetching poem from server.
Canceling poem download.你應該看到慢速服務器已經(jīng)停止了輸出打印它所發(fā)送的詩歌片段,因為我們的代理斷開了鏈接.
你可以多次啟動和停止客戶端來證實每個下載每次都被取消了.但是如果你讓整首詩運行完,那么代理將緩存它并且在此之后立即發(fā)送它.
以上我們曾不止一次說取消一個已經(jīng)激發(fā)的 deferred 是沒有效果的.然而,這不是十分正確.在 第十三部分 中,我們知道 deferred 的回調(diào)和錯誤回調(diào)也可能返回另一個 deferred.在那種情況下,原始的(外層) deferred 暫停執(zhí)行它的回調(diào)鏈并且等待內(nèi)層 deferred 激發(fā)(參見 圖28).
如此, 即使一個 deferred 激發(fā)了發(fā)出異步請求的高層代碼,它也不能接收到結果,因為在等待內(nèi)層 deferred 完成之前回調(diào)鏈暫停了. 所以當高層代碼取消這個外部 deferred 時會發(fā)生什么情況呢? 在這種情況下,外部 deferred 不僅僅是取消它自己(它已經(jīng)激發(fā)了);相反地,這個 deferred 也會取消內(nèi)部的 deferred.
所以當你取消一個 deferred 時,你可能不是在取消主異步操作,而是一些其他的作為前者結果所觸發(fā)的異步操作.呼!
我們可以用一個例子來說明.考慮代碼 deferred-cancel/defer-cancel-12.py
from twisted.internet import defer
def cancel_outer(d):
print "outer cancel callback."
def cancel_inner(d):
print "inner cancel callback."
def first_outer_callback(res):
print 'first outer callback, returning inner deferred'
return inner_d
def second_outer_callback(res):
print 'second outer callback got:', res
def outer_errback(err):
print 'outer errback got:', err
outer_d = defer.Deferred(cancel_outer)
inner_d = defer.Deferred(cancel_inner)
outer_d.addCallback(first_outer_callback)
outer_d.addCallbacks(second_outer_callback, outer_errback)
outer_d.callback('result')
# at this point the outer deferred has fired, but is paused
# on the inner deferred.
print 'canceling outer deferred.'
outer_d.cancel()
print 'done'在這個例子中,我們創(chuàng)建了兩個 deferred, outer 和 inner,并且有一個外部回調(diào)返回內(nèi)部的 deferred. 首先,我們激發(fā)外部 deferred,然后取消它. 輸出結果如下:
first outer callback, returning inner deferred
canceling outer deferred.
inner cancel callback.
outer errback got: [Failure instance: Traceback (failure with no frames):
<class 'twisted.internet.defer.CancelledError'>: ]
done正如你看到的,取消外部 deferred 并沒有使外部 cancel 回調(diào)被激發(fā). 相反,它取消了內(nèi)部 deferred,所以內(nèi)部 cancel 回調(diào)被激發(fā)了,之后外部錯誤回調(diào)收到 CancelledError (來自內(nèi)部 deferred).
你可能需要仔細看一看那些代碼,并且做些變化看看如何影響結果.
取消 deferred 是非常有用的操作,使我們的程序避免去做不需要的工作. 然而正如我們看到的,它可能有一點點棘手.
需要明白的一個重要事實是取消一個 deferred 并不意味著取消了它后面的異步操作.事實上,當寫這篇文章時,很多 deferreds 并不會被真的"取消",因為大部分Twisted代碼寫于Twisted 10.1.0之前并且還沒有被升級.這包括很多Twisted本身的APIs！檢查文檔或源代碼去發(fā)現(xiàn)"取消 deferred"是否真的取消了背后的請求,還是僅僅忽略它.
第二個重要事實是從你的異步APIs返回的 deferred 并不一定在完整意義上可取消. 如果你希望在自己的程序中實現(xiàn)取消,你應該先研究一下Twisted源代碼中的許多例子. Cancellation 是一個嶄新的特性,所以它的模式和最好實踐還在制定當中.
現(xiàn)在我們已經(jīng)學習了關于 Deferreds 的方方面面以及Twisted背后的核心概念. 這意味著我們沒什么需要介紹的了,因為Twisted的其余部分主要包括一些特定的應用,如網(wǎng)絡編程或異步數(shù)據(jù)庫處理.故而,在 接下來兩部分,我們想走點彎路,看看其他兩個使用異步I/O的系統(tǒng)跟Twisted有何理念相似之處.之后,在尾聲中,我們會總結并且建議一些幫助你繼續(xù)學習Twisted的方法.
cancelled"嗎? 真的 這取決于你的心情.cancellation 的實現(xiàn).deferred 的例子.研究它們的實現(xiàn).get_poetry 方法返回的 deferred, 使其可取消.reactor 的例子展示取消外部 deferred,它被內(nèi)層 deferred 暫停了.如果使用 callLater 你需要小心選擇延遲時間,以確保外層 deferred 在正確的時刻被取消.本部分原作參見: dave @ http://krondo.com/blog/?p=2601
本部分翻譯內(nèi)容參見luocheng @ https://github.com/luocheng/twisted-intro-cn/blob/master/p19.rst
掃碼關注鍍金池微信號
手機訂課，優(yōu)惠多多