日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

[[377099]]

為云岡等地區(qū)用戶提供了全套網(wǎng)頁設計制作服務，及云岡網(wǎng)站建設行業(yè)解決方案。主營業(yè)務為成都網(wǎng)站建設、做網(wǎng)站、云岡網(wǎng)站設計，以傳統(tǒng)方式定制建設網(wǎng)站，并提供域名空間備案等一條龍服務，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務。我們深信只要達到每一位用戶的要求，就會得到認可，從而選擇與我們長期合作。這樣，我們也可以走得更遠！

本文轉(zhuǎn)載自微信公眾號「老王Plus」，作者老王Plus的老王 。轉(zhuǎn)載本文請聯(lián)系老王Plus公眾號。

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

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

話題本身跟gRPC沒有太大關(guān)系。應用中，我用到了全雙工數(shù)據(jù)管道這樣一個相對復雜的概念。

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

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

一、開始

假設我們有這樣一個API：

• 客戶端建立連接
• 有一個SendAsync消息從客戶端發(fā)送到服務器
• 有一個TryReceiveAsync消息，試圖等待來自服務器的消息(服務器有消息發(fā)送為True，返之為False)
• 服務器控制數(shù)據(jù)流終止，如果服務器發(fā)送完最后一條消息，則客戶端不再發(fā)送任何消息。

接口代碼可以寫成這樣：

 
 
 

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

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

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

 
 
 

  
  
  
ITransport transport; 
  
  
  
public async IAsyncEnumerable ReceiveAsync([EnumeratorCancellation] CancellationToken cancellationToken) 
  
  
  
{ 
  
  
  
    while (true) 
  
  
  
    { 
  
  
  
        var (success, message) = 
  
  
  
            await transport.TryReceiveAsync(cancellationToken); 
  
  
  
        if (!success) break; 
  
  
  
        yield return message; 
  
  
  
    } 
  
  
  
} 
  
  
  
 
  
  
  
public async ValueTask SendAsync(IAsyncEnumerable data, CancellationToken cancellationToken) 
  
  
  
{ 
  
  
  
    await foreach (var message in data.WithCancellation(cancellationToken)) 
  
  
  
    { 
  
  
  
        await transport.SendAsync(message, cancellationToken); 
  
  
  
    } 
  
  
  
} 
 
 
 

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

二、解決終止標志

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

真的做好了嗎?

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

實際上，會有比我們想像中更多的終止路徑：

• 我們可能已經(jīng)為這兩個方法提供了一個外部的終止令牌，并且這個令牌可能已經(jīng)被觸發(fā)
• ReceiveAsync的消費者可能已經(jīng)通過WithCancellation提供了一個終止令牌給GetAsyncEnumerator，并且這個令牌可能已經(jīng)被觸發(fā)
• 我們的發(fā)送/接收代碼可能出錯了
• ReceiveAsync的消費者在數(shù)據(jù)獲取到中途，要終止獲取了 - 一個簡單的原因是處理收到的數(shù)據(jù)時出錯了
• SendAsync中的生產(chǎn)者可能發(fā)生了錯誤

這只是一些可能的例子，但實際的可能會更多。

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

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

 
 
 

  
  
  
public IAsyncEnumerable Duplex(IAsyncEnumerable request, CancellationToken cancellationToken = default); 
 
 
 

這個方法：

• 允許它傳入一個生產(chǎn)者
• 通話它傳入一個外部的終止令牌
• 有一個異步的響應返回

使用時，我們可以這樣做：

 
 
 

  
  
  
await foreach (MyResponse item in client.Duplex(ProducerAsync())) 
  
  
  
{ 
  
  
  
    // ... todo 
  
  
  
} 
  
  
  
async IAsyncEnumerable ProducerAsync([EnumeratorCancellation] CancellationToken cancellationToken = default) 
  
  
  
{ 
  
  
  
    for (int i = 0; i < 100; i++) 
  
  
  
    { 
  
  
  
        yield return new MyRequest(i); 
  
  
  
        await Task.Delay(100, cancellationToken); 
  
  
  
    } 
  
  
  
} 
 
 
 

上面這段代碼中，我們ProducerAsync還沒有實現(xiàn)太多內(nèi)容，目前只是傳遞了一個占位符。稍后我們可以枚舉它，而枚舉行為實際上調(diào)用了代碼。

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

• 通過cancellationToken傳入的外部令牌
• 使用過程中可能傳遞給GetAsyncEnumerator()的潛在的令牌

這兒，為什么不是之前列出的更多種終止方式呢?這兒要考慮到編譯器的組合方式。我們需要的不是一個CancellationToken，而是一個CancellationTokenSource。

 
 
 

  
  
  
public IAsyncEnumerable Duplex(IAsyncEnumerable request, CancellationToken cancellationToken = default) => DuplexImpl(transport, request, cancellationToken); 
  
  
  
 
  
  
  
private async static IAsyncEnumerable DuplexImpl(ITransport transport, IAsyncEnumerable request, CancellationToken externalToken, [EnumeratorCancellation] CancellationToken enumeratorToken = default) 
  
  
  
{ 
  
  
  
    using var allDone = CancellationTokenSource.CreateLinkedTokenSource(externalToken, enumeratorToken); 
  
  
  
    // ... todo 
  
  
  
} 
 
 
 

這里，DuplexImpl方法允許枚舉終止，但又與外部終止標記保持分離。這樣，在編譯器層面不會被合并。在里面，CreateLinkedTokenSource反倒像編譯器的處理。

現(xiàn)在，我們有一個CancellationTokenSource，需要時，我們可能通過它來終止循環(huán)的運行。

 
 
 

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

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

下一個是生產(chǎn)者，也就是SendAsync。它也是雙工的，對傳入的消息沒有影響，所以可以用Task.Run作為一個獨立的代碼路徑開始運行，而如果生產(chǎn)者出現(xiàn)錯誤，則終止發(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; 
 
 
 

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

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

 
 
 

  
  
  
while (true) 
  
  
  
{ 
  
  
  
    var (success, message) = await transport.TryReceiveAsync(allDone.Token); 
  
  
  
    if (!success) break; 
  
  
  
    yield return message; 
  
  
  
} 
  
  
  
 
  
  
  
allDone.Cancel(); 
 
 
 

最后，把這部分代碼合在一起看看：

 
 
 

  
  
  
private async static IAsyncEnumerable DuplexImpl(ITransport transport, IAsyncEnumerable 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)的處理就這么多。這里實現(xiàn)的關(guān)鍵點是：

• 外部令牌和枚舉器令牌都對allDone有貢獻
• 傳輸中發(fā)送和接收代碼使用allDone.Token
• 生產(chǎn)者枚舉使用allDone.Token
• 任何情況下退出枚舉器，allDone都會被終止
• 如果傳輸接收錯誤，則allDone被終止
• 如果消費者提前終止，則allDone被終止
• 當我們收到來自服務器的最后一條消息后，allDone被終止
• 如果生產(chǎn)者或傳輸發(fā)送錯誤，allDone被終止

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

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

 

分享標題：開發(fā)進階：DotnetCore多路徑異步終止
分享地址：http://m.5511xx.com/article/ccdcpdg.html