.NET中使用异步可枚举同时读写gRPC全双工通道方案咨询

你当前实现的核心缺陷是请求写入和响应读取完全串行执行：所有输入数据写完之前，响应流的读取逻辑不会启动。gRPC底层的接收缓冲区被服务端提前返回的结果打满后，就会抛出OperationCancelled异常。
你提到的单循环交替读写模式不可用，本质是因为ResponseStream.MoveNext()是阻塞式等待调用，和请求写入的时序无法对齐，容易触发死锁。

使用System.Threading.Channels做读写并行桥接即可，不需要修改服务端逻辑，也完全不破坏上层的IAsyncEnumerable调用模式，核心思路是把gRPC响应读取、gRPC请求写入拆分为两个独立并行的异步任务，响应读取任务在调用初始化后立刻启动，从根源避免缓冲区满的问题。
首先引入命名空间：
using System.Threading.Channels;
具体实现代码如下：

public async IAsyncEnumerable<Result> Results(
    IAsyncEnumerable<Input> inputs, 
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    // 初始化本地异步队列做结果缓存
    var resultChannel = Channel.CreateUnbounded<Result>(new UnboundedChannelOptions
    {
        SingleReader = true,
        SingleWriter = true
    });
    
    // 初始化gRPC双向流
    var duplex = Server.GetResults(cancellationToken: cancellationToken);

    // 任务1：启动即开始持续读取gRPC响应，写入本地队列，不会等待请求全部写完
    var readTask = Task.Run(async () =>
    {
        try
        {
            await foreach (var res in duplex.ResponseStream.ToAsyncEnumerable(cancellationToken))
            {
                await resultChannel.Writer.WriteAsync(res, cancellationToken);
            }
            resultChannel.Writer.Complete();
        }
        catch (Exception ex)
        {
            resultChannel.Writer.Complete(ex);
        }
    }, cancellationToken);

    // 任务2：独立执行请求写入逻辑
    var writeTask = Task.Run(async () =>
    {
        try
        {
            await foreach (var input in inputs.WithCancellation(cancellationToken))
            {
                await duplex.RequestStream.WriteAsync(input, cancellationToken);
            }
            await duplex.RequestStream.CompleteAsync();
        }
        catch (Exception ex)
        {
            resultChannel.Writer.TryComplete(ex);
            duplex.Dispose();
        }
    }, cancellationToken);

    // 直接从本地队列读取结果返回，完全保留yield写法
    await foreach (var result in resultChannel.Reader.ReadAllAsync(cancellationToken))
    {
        yield return result;
    }

    // 等待所有任务执行完成，抛出未处理异常
    await Task.WhenAll(readTask, writeTask);
}

方案优势 #

• 对外接口完全保持IAsyncEnumerable<Result>返回类型，上层调用逻辑不需要任何修改，yield return的使用方式也完全保留
• 响应读取逻辑在gRPC流初始化后立刻启动，不会出现请求写完才开始读导致缓冲区打满的问题
• 读写逻辑运行在独立异步任务中，没有使用你提到的单循环交替读写模式，不会出现MoveNext()阻塞导致写入停滞的死锁问题
• 不需要依赖AsyncDuplexStreamingCall未提供的IsDataAvailable类检测接口，Channel本身的异步读写机制天然处理了数据就绪的通知逻辑

• 如果担心本地队列内存占用过高，可以将无界Channel替换为有界Channel，配置BoundedChannelOptions设置合理容量，将FullMode设置为BoundedChannelFullMode.Wait即可，不会出现数据丢失问题
• 必须将枚举器的CancellationToken透传到所有异步操作中，调用方发起取消时可以立刻终止所有读写任务与gRPC调用
• 你提到的4级串联流水线场景可以直接复用该写法，每一层的Channel桥接开销极低，不会成为性能瓶颈

如果无法引入Channel依赖，也可以基于TaskCompletionSource实现轻量生产消费队列完成同样的并行桥接逻辑，但异常处理和边界场景处理复杂度远高于Channel方案，无特殊需求优先使用上述Channel实现。

内容的提问来源于stack exchange，提问作者Daniel Leiszen