今天用一個簡單例子說說異步的多路徑終止，我盡可能寫得容易理解吧，但今天的內容需要有一定的編程能力，

今天這個話題，來自于最近對gRPC的一些技術研究，

話題本身跟gRPC沒有太大關系，應用中，我用到了全雙工資料管道這樣一個相對復雜的概念，

我們知道，全雙工連接是兩個節點之間的連接，但不是簡單的“請求-回應”連接，任何一個節點都可以在任何時間發送訊息，概念上，還是有客戶端和服務端的區分，但這僅僅是概念上，只是為了區分誰在監聽連接嘗試，誰在建立連接，實際上，做一個雙工的API比做一個“請求-回應”式的API要復雜得多，

由此，延伸出了另一個想法：做個類別庫，在庫內部構建雙工管道，供給消費者時，只暴露簡單的內容和熟悉的方式，

為了防止不提供原網址的轉載，特在這里加上原文鏈接：https://www.cnblogs.com/tiger-wang/p/14297970.html

一、開始

假設我們有這樣一個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: receive

await 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，除非你知道你在做什么，
?

微信公眾號：老王Plus

掃描二維碼，關注個人公眾號，可以第一時間得到最新的個人文章和內容推送

本文著作權歸作者所有，轉載請保留此宣告和原文鏈接

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/251289.html

標籤：.NET技术

上一篇：Core3.1 微信v3 JSAPI支付

下一篇：EFCore 5 新特性 —— Savepoints

開發進階：Dotnet Core多路徑異步終止

一、開始

二、解決終止標志

三、總結