今天用一个简单例子说说异步的多路径终止。我尽可能写得容易理解吧，但今天的内容需要有一定的编程能力。
今天这个话题，来自于最近对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 }