开发进阶:Dotnet Core多路径异步终止
今天用一個簡單例子說說異步的多路徑終止。我盡可能寫得容易理解吧,但今天的內容需要有一定的編程能力。
?
今天這個話題,來自于最近對gRPC的一些技術研究。
話題本身跟gRPC沒有太大關系。應用中,我用到了全雙工數據管道這樣一個相對復雜的概念。
我們知道,全雙工連接是兩個節點之間的連接,但不是簡單的“請求-響應”連接。任何一個節點都可以在任何時間發送消息。概念上,還是有客戶端和服務端的區分,但這僅僅是概念上,只是為了區分誰在監聽連接嘗試,誰在建立連接。實際上,做一個雙工的API比做一個“請求-響應”式的API要復雜得多。
由此,延伸出了另一個想法:做個類庫,在庫內部構建雙工管道,供給消費者時,只暴露簡單的內容和熟悉的方式。
?
一、開始
假設我們有這樣一個API:
客戶端建立連接
有一個SendAsync消息從客戶端發送到服務器
有一個TryReceiveAsync消息,試圖等待來自服務器的消息(服務器有消息發送為True,返之為False)
服務器控制數據流終止,如果服務器發送完最后一條消息,則客戶端不再發送任何消息。
?
接口代碼可以寫成這樣:
interface?ITransport<TRequest,?TResponse>?:?IAsyncDisposable {ValueTask?SendAsync(TRequest?request,?CancellationToken?cancellationToken);ValueTask<(bool?Success,?TResponse?Message)>?TryReceiveAsync(CancellationToken?cancellationToken); }忽略連接的部分,代碼看起來并不復雜。
下面,我們創建兩個循環,并通過枚舉器公開數據:
ITransport<TRequest,?TResponse>?transport; public?async?IAsyncEnumerable<TResponse>?ReceiveAsync([EnumeratorCancellation]?CancellationToken?cancellationToken) {while?(true){var?(success,?message)?=await?transport.TryReceiveAsync(cancellationToken);if?(!success)?break;yield?return?message;} }public?async?ValueTask?SendAsync(IAsyncEnumerable<TRequest>?data,?CancellationToken?cancellationToken) {await?foreach?(var?message?in?data.WithCancellation(cancellationToken)){await?transport.SendAsync(message,?cancellationToken);} }這里面用到了異步迭代器相關的概念。如果不明白,可以去看我的另一篇專門討論異步迭代器的文章,【傳送門】。
?
二、解決終止標志
好像做好了,我們用循環接收和發送,并傳遞了外部的終止標志給這兩個方法。
真的做好了嗎?
還沒有。問題出在終止標志上。我們沒有考慮到這兩個流是相互依賴的,特別是,我們不希望生產者(使用SendAsync的代碼)在任何連接失敗的場景中仍然運行。
實際上,會有比我們想像中更多的終止路徑:
我們可能已經為這兩個方法提供了一個外部的終止令牌,并且這個令牌可能已經被觸發
ReceiveAsync的消費者可能已經通過WithCancellation提供了一個終止令牌給GetAsyncEnumerator,并且這個令牌可能已經被觸發
我們的發送/接收代碼可能出錯了
ReceiveAsync的消費者在數據獲取到中途,要終止獲取了 - 一個簡單的原因是處理收到的數據時出錯了
SendAsync中的生產者可能發生了錯誤
這只是一些可能的例子,但實際的可能會更多。
本質上,這些都表示連接終止,因此我們需要以某種方式包含所有這些場景,進而允許發送和接收路徑之間傳達問題。換句話說,我們需要自己的CancellationTokenSource。
?
顯然,這種需求,用庫來解決是比較完美的。我們可以把這些復雜的內容放在一個消費者可以訪問的單一API中:
public?IAsyncEnumerable<TResponse>?Duplex(IAsyncEnumerable<TRequest>?request,?CancellationToken?cancellationToken?=?default);這個方法:
允許它傳入一個生產者
通話它傳入一個外部的終止令牌
有一個異步的響應返回
使用時,我們可以這樣做:
await?foreach?(MyResponse?item?in?client.Duplex(ProducerAsync())) {//?...?todo } async?IAsyncEnumerable<MyRequest>?ProducerAsync([EnumeratorCancellation]?CancellationToken?cancellationToken?=?default) {for?(int?i?=?0;?i?<?100;?i++){yield?return?new?MyRequest(i);await?Task.Delay(100,?cancellationToken);} }?
上面這段代碼中,我們ProducerAsync還沒有實現太多內容,目前只是傳遞了一個占位符。稍后我們可以枚舉它,而枚舉行為實際上調用了代碼。
回到Duplex。這個方法,至少需要考慮兩種不同的終止方式:
通過cancellationToken傳入的外部令牌
使用過程中可能傳遞給GetAsyncEnumerator()的潛在的令牌
這兒,為什么不是之前列出的更多種終止方式呢?這兒要考慮到編譯器的組合方式。我們需要的不是一個CancellationToken,而是一個CancellationTokenSource。
public?IAsyncEnumerable<TResponse>?Duplex(IAsyncEnumerable<TRequest>?request,?CancellationToken?cancellationToken?=?default)?=>?DuplexImpl(transport,?request,?cancellationToken);private?async?static?IAsyncEnumerable<TResponse>?DuplexImpl(ITransport<TRequest,?TResponse>?transport,?IAsyncEnumerable<TRequest>?request,?CancellationToken?externalToken,?[EnumeratorCancellation]?CancellationToken?enumeratorToken?=?default) {using?var?allDone?=?CancellationTokenSource.CreateLinkedTokenSource(externalToken,?enumeratorToken);//?...?todo }這里,DuplexImpl方法允許枚舉終止,但又與外部終止標記保持分離。這樣,在編譯器層面不會被合并。在里面,CreateLinkedTokenSource反倒像編譯器的處理。
現在,我們有一個CancellationTokenSource,需要時,我們可能通過它來終止循環的運行。
using?var?allDone?=?CancellationTokenSource.CreateLinkedTokenSource(externalToken,?enumeratorToken); try {//?...?todo } finally {allDone.Cancel(); }通過這種方式,我們可以處理這樣的場景:消費者沒有獲取所有數據,而我們想要觸發allDone,但是我們退出了DuplexImpl。這時候,迭代器的作用就很大了,它讓程序變得更簡單,因為用了using,最終里面的任何內容都會定位到Dispose/DisposeAsync。
?
下一個是生產者,也就是SendAsync。它也是雙工的,對傳入的消息沒有影響,所以可以用Task.Run作為一個獨立的代碼路徑開始運行,而如果生產者出現錯誤,則終止發送。上邊的todo部分,可以加入:
var?send?=?Task.Run(async?()?=> {try{await?foreach?(var?message?in?request.WithCancellation(allDone.Token)){await?transport.SendAsync(message,?allDone.Token);}}catch{allDone.Cancel();throw;} },?allDone.Token);//?...?todo:?receiveawait?send;這里啟動了一個生產者的并行操作SendAsync。注意,這里我們用標記allDone.Token把組合的終止標記傳遞給生產者。延遲await是為了允許ProducerAsync方法里可以使用終止令牌,以滿足復合雙工操作的生命周期要求。
?
這樣,接收代碼就變成了:
while?(true) {var?(success,?message)?=?await?transport.TryReceiveAsync(allDone.Token);if?(!success)?break;yield?return?message; }allDone.Cancel();?
最后,把這部分代碼合在一起看看:
private?async?static?IAsyncEnumerable<TResponse>?DuplexImpl(ITransport<TRequest,?TResponse>?transport,?IAsyncEnumerable<TRequest>?request,?CancellationToken?externalToken,?[EnumeratorCancellation]?CancellationToken?enumeratorToken?=?default) {using?var?allDone?=?CancellationTokenSource.CreateLinkedTokenSource(externalToken,?enumeratorToken);try{var?send?=?Task.Run(async?()?=>{try{await?foreach?(var?message?in?request.WithCancellation(allDone.Token)){await?transport.SendAsync(message,?allDone.Token);}}catch{allDone.Cancel();throw;}},?allDone.Token);while?(true){var?(success,?message)?=?await?transport.TryReceiveAsync(allDone.Token);if?(!success)?break;yield?return?message;}allDone.Cancel();await?send;}finally{allDone.Cancel();} }?
三、總結
相關的處理就這么多。這里實現的關鍵點是:
外部令牌和枚舉器令牌都對allDone有貢獻
傳輸中發送和接收代碼使用allDone.Token
生產者枚舉使用allDone.Token
任何情況下退出枚舉器,allDone都會被終止
如果傳輸接收錯誤,則allDone被終止
如果消費者提前終止,則allDone被終止
當我們收到來自服務器的最后一條消息后,allDone被終止
如果生產者或傳輸發送錯誤,allDone被終止
?
最后多說一點,關于ConfigureAwait(false):
默認情況下,await包含一個對SynchronizationContext.Current的檢查。除了表示額外的上下文切換之外,在UI應用程序的情況下,它也意味著在UI線程上運行不需要在UI線程上運行的代碼。庫代碼通常不需要這樣做。因此,在庫代碼中,通常應該在所有用到await的地方使用. configureawait (false)來繞過這個檢查。而在一般應用程序的代碼中,應該默認只使用await而不使用ConfigureAwait,除非你知道你在做什么。
喜歡就來個三連,讓更多人因你而受益
總結
以上是生活随笔為你收集整理的开发进阶:Dotnet Core多路径异步终止的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 并行模型Actor
- 下一篇: 跨平台导PDF,结合wkhtmltopd