今天用一個(gè)簡(jiǎn)單例子說(shuō)說(shuō)異步的多路徑終止。我盡可能寫(xiě)得容易理解吧，但今天的內(nèi)容需要有一定的編程能力。

?

今天這個(gè)話(huà)題，來(lái)自于最近對(duì)gRPC的一些技術(shù)研究。

話(huà)題本身跟gRPC沒(méi)有太大關(guān)系。應(yīng)用中，我用到了全雙工數(shù)據(jù)管道這樣一個(gè)相對(duì)復(fù)雜的概念。

我們知道，全雙工連接是兩個(gè)節(jié)點(diǎn)之間的連接，但不是簡(jiǎn)單的“請(qǐng)求-響應(yīng)”連接。任何一個(gè)節(jié)點(diǎn)都可以在任何時(shí)間發(fā)送消息。概念上，還是有客戶(hù)端和服務(wù)端的區(qū)分，但這僅僅是概念上，只是為了區(qū)分誰(shuí)在監(jiān)聽(tīng)連接嘗試，誰(shuí)在建立連接。實(shí)際上，做一個(gè)雙工的API比做一個(gè)“請(qǐng)求-響應(yīng)”式的API要復(fù)雜得多。

由此，延伸出了另一個(gè)想法：做個(gè)類(lèi)庫(kù)，在庫(kù)內(nèi)部構(gòu)建雙工管道，供給消費(fèi)者時(shí)，只暴露簡(jiǎn)單的內(nèi)容和熟悉的方式。

?

一、開(kāi)始

假設(shè)我們有這樣一個(gè)API：

• 客戶(hù)端建立連接

• 有一個(gè)SendAsync消息從客戶(hù)端發(fā)送到服務(wù)器

• 有一個(gè)TryReceiveAsync消息，試圖等待來(lái)自服務(wù)器的消息（服務(wù)器有消息發(fā)送為T(mén)rue，返之為False）

• 服務(wù)器控制數(shù)據(jù)流終止，如果服務(wù)器發(fā)送完最后一條消息，則客戶(hù)端不再發(fā)送任何消息。

?

接口代碼可以寫(xiě)成這樣：

interface?ITransport<TRequest,?TResponse>?:?IAsyncDisposable {ValueTask?SendAsync(TRequest?request,?CancellationToken?cancellationToken);ValueTask<(bool?Success,?TResponse?Message)>?TryReceiveAsync(CancellationToken?cancellationToken); }

忽略連接的部分，代碼看起來(lái)并不復(fù)雜。

下面，我們創(chuàng)建兩個(gè)循環(huán)，并通過(guò)枚舉器公開(kāi)數(shù)據(jù)：

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);} }

這里面用到了異步迭代器相關(guān)的概念。如果不明白，可以去看我的另一篇專(zhuān)門(mén)討論異步迭代器的文章，【傳送門(mén)】。

?

二、解決終止標(biāo)志

好像做好了，我們用循環(huán)接收和發(fā)送，并傳遞了外部的終止標(biāo)志給這兩個(gè)方法。

真的做好了嗎？

還沒(méi)有。問(wèn)題出在終止標(biāo)志上。我們沒(méi)有考慮到這兩個(gè)流是相互依賴(lài)的，特別是，我們不希望生產(chǎn)者（使用SendAsync的代碼）在任何連接失敗的場(chǎng)景中仍然運(yùn)行。

實(shí)際上，會(huì)有比我們想像中更多的終止路徑：

• 我們可能已經(jīng)為這兩個(gè)方法提供了一個(gè)外部的終止令牌，并且這個(gè)令牌可能已經(jīng)被觸發(fā)

• ReceiveAsync的消費(fèi)者可能已經(jīng)通過(guò)WithCancellation提供了一個(gè)終止令牌給GetAsyncEnumerator，并且這個(gè)令牌可能已經(jīng)被觸發(fā)

• 我們的發(fā)送/接收代碼可能出錯(cuò)了

• ReceiveAsync的消費(fèi)者在數(shù)據(jù)獲取到中途，要終止獲取了 - 一個(gè)簡(jiǎn)單的原因是處理收到的數(shù)據(jù)時(shí)出錯(cuò)了

• SendAsync中的生產(chǎn)者可能發(fā)生了錯(cuò)誤

這只是一些可能的例子，但實(shí)際的可能會(huì)更多。

本質(zhì)上，這些都表示連接終止，因此我們需要以某種方式包含所有這些場(chǎng)景，進(jìn)而允許發(fā)送和接收路徑之間傳達(dá)問(wèn)題。換句話(huà)說(shuō)，我們需要自己的CancellationTokenSource。

?

顯然，這種需求，用庫(kù)來(lái)解決是比較完美的。我們可以把這些復(fù)雜的內(nèi)容放在一個(gè)消費(fèi)者可以訪(fǎng)問(wèn)的單一API中：

public?IAsyncEnumerable<TResponse>?Duplex(IAsyncEnumerable<TRequest>?request,?CancellationToken?cancellationToken?=?default);

這個(gè)方法：

• 允許它傳入一個(gè)生產(chǎn)者

• 通話(huà)它傳入一個(gè)外部的終止令牌

• 有一個(gè)異步的響應(yīng)返回

使用時(shí)，我們可以這樣做：

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還沒(méi)有實(shí)現(xiàn)太多內(nèi)容，目前只是傳遞了一個(gè)占位符。稍后我們可以枚舉它，而枚舉行為實(shí)際上調(diào)用了代碼。

回到Duplex。這個(gè)方法，至少需要考慮兩種不同的終止方式：

• 通過(guò)cancellationToken傳入的外部令牌

• 使用過(guò)程中可能傳遞給GetAsyncEnumerator()的潛在的令牌

這兒，為什么不是之前列出的更多種終止方式呢？這兒要考慮到編譯器的組合方式。我們需要的不是一個(gè)CancellationToken，而是一個(gè)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方法允許枚舉終止，但又與外部終止標(biāo)記保持分離。這樣，在編譯器層面不會(huì)被合并。在里面，CreateLinkedTokenSource反倒像編譯器的處理。

現(xiàn)在，我們有一個(gè)CancellationTokenSource，需要時(shí)，我們可能通過(guò)它來(lái)終止循環(huán)的運(yùn)行。

using?var?allDone?=?CancellationTokenSource.CreateLinkedTokenSource(externalToken,?enumeratorToken); try {//?...?todo } finally {allDone.Cancel(); }

通過(guò)這種方式，我們可以處理這樣的場(chǎng)景：消費(fèi)者沒(méi)有獲取所有數(shù)據(jù)，而我們想要觸發(fā)allDone，但是我們退出了DuplexImpl。這時(shí)候，迭代器的作用就很大了，它讓程序變得更簡(jiǎn)單，因?yàn)橛昧藆sing，最終里面的任何內(nèi)容都會(huì)定位到Dispose/DisposeAsync。

?

下一個(gè)是生產(chǎn)者，也就是SendAsync。它也是雙工的，對(duì)傳入的消息沒(méi)有影響，所以可以用Task.Run作為一個(gè)獨(dú)立的代碼路徑開(kāi)始運(yùn)行，而如果生產(chǎn)者出現(xiàn)錯(cuò)誤，則終止發(fā)送。上邊的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;

這里啟動(dòng)了一個(gè)生產(chǎn)者的并行操作SendAsync。注意，這里我們用標(biāo)記allDone.Token把組合的終止標(biāo)記傳遞給生產(chǎn)者。延遲await是為了允許ProducerAsync方法里可以使用終止令牌，以滿(mǎn)足復(fù)合雙工操作的生命周期要求。

?

這樣，接收代碼就變成了：

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();} }

?

三、總結(jié)

相關(guān)的處理就這么多。這里實(shí)現(xiàn)的關(guān)鍵點(diǎn)是：

• 外部令牌和枚舉器令牌都對(duì)allDone有貢獻(xiàn)

• 傳輸中發(fā)送和接收代碼使用allDone.Token

• 生產(chǎn)者枚舉使用allDone.Token

• 任何情況下退出枚舉器，allDone都會(huì)被終止

• 如果傳輸接收錯(cuò)誤，則allDone被終止

• 如果消費(fèi)者提前終止，則allDone被終止

• 當(dāng)我們收到來(lái)自服務(wù)器的最后一條消息后，allDone被終止

• 如果生產(chǎn)者或傳輸發(fā)送錯(cuò)誤，allDone被終止

?

最后多說(shuō)一點(diǎn)，關(guān)于ConfigureAwait(false)：

默認(rèn)情況下，await包含一個(gè)對(duì)SynchronizationContext.Current的檢查。除了表示額外的上下文切換之外，在UI應(yīng)用程序的情況下，它也意味著在UI線(xiàn)程上運(yùn)行不需要在UI線(xiàn)程上運(yùn)行的代碼。庫(kù)代碼通常不需要這樣做。因此，在庫(kù)代碼中，通常應(yīng)該在所有用到await的地方使用. configureawait (false)來(lái)繞過(guò)這個(gè)檢查。而在一般應(yīng)用程序的代碼中，應(yīng)該默認(rèn)只使用await而不使用ConfigureAwait，除非你知道你在做什么。

喜歡就來(lái)個(gè)三連，讓更多人因你而受益

總結(jié)

以上是生活随笔為你收集整理的开发进阶：Dotnet Core多路径异步终止的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。