如何理解Dotnet Core多路径异步终止

本篇内容介绍了“如何理解Dotnet Core多路径异步终止”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

一、开始

假设我们有这样一个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();     } }

“如何理解Dotnet Core多路径异步终止”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站，小编将为大家输出更多高质量的实用文章！