如何理解Dotnet Core多路徑異步終止

本篇內(nèi)容介紹了“如何理解Dotnet Core多路徑異步終止”的有關(guān)知識(shí)，在實(shí)際案例的操作過程中，不少人都會(huì)遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

一、開始

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

• 客戶端建立連接

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

• 有一個(gè)TryReceiveAsync消息，試圖等待來自服務(wù)器的消息(服務(wù)器有消息發(fā)送為True，返之為False)

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

接口代碼可以寫成這樣：

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

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

下面，我們創(chuàng)建兩個(gè)循環(huán)，并通過枚舉器公開數(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)的概念。如果不明白，可以去看我的另一篇專門討論異步迭代器的文章，【傳送門】。

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

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

真的做好了嗎?

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

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

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

• ReceiveAsync的消費(fèi)者可能已經(jīng)通過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á)問題。換句話說，我們需要自己的CancellationTokenSource。

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

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

這個(gè)方法：

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

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

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

• 通過cancellationToken傳入的外部令牌

• 使用過程中可能傳遞給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í)，我們可能通過它來終止循環(huán)的運(yùn)行。

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

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

下一個(gè)是生產(chǎn)者，也就是SendAsync。它也是雙工的，對(duì)傳入的消息沒有影響，所以可以用Task.Run作為一個(gè)獨(dú)立的代碼路徑開始運(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: receive  await send;

這里啟動(dòng)了一個(gè)生產(chǎn)者的并行操作SendAsync。注意，這里我們用標(biāo)記allDone.Token把組合的終止標(biāo)記傳遞給生產(chǎn)者。延遲await是為了允許ProducerAsync方法里可以使用終止令牌，以滿足復(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();     } }

“如何理解Dotnet Core多路徑異步終止”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！