.NET5下基于protobuf-net实现无循环无延迟Code First gRPC服务端流

完全可以实现，核心是用.NET 内置的Channel<T>作为事件触发逻辑与IAsyncEnumerable<T>返回值之间的异步桥接层，不需要在接口定义中暴露服务端专属的流参数，也不需要轮询循环，就能做到事件触发后即时推送。

第一步：定义服务契约 #

Code First模式下服务端流的方法只需要返回IAsyncEnumerable<T>即可，不需要引入任何服务端专属类型，客户端可正常调用：

[Service]
public interface IStockTickerService
{
    // 入参为订阅请求参数，返回IAsyncEnumerable对应服务端流
    IAsyncEnumerable<StockUpdate> Subscribe(SubscribeRequest request, CallContext context = default);
}

第二步：实现无循环即时推送逻辑 #

用Channel作为事件的中转容器，事件触发时往Channel写入数据，IAsyncEnumerable直接读取Channel的输出流即可：

public class StockTickerService : IStockTickerService
{
    private readonly ISubscriberFactory _subscriberFactory;
    private readonly ILogger<StockTickerService> _logger;

    public StockTickerService(ISubscriberFactory subscriberFactory, ILogger<StockTickerService> logger)
    {
        _subscriberFactory = subscriberFactory;
        _logger = logger;
    }

    public async IAsyncEnumerable<StockUpdate> Subscribe(SubscribeRequest request, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // 初始化无界Channel，配置为单读单写模式性能最优
        var channel = Channel.CreateUnbounded<StockUpdate>(new UnboundedChannelOptions
        {
            SingleReader = true,
            SingleWriter = true,
            AllowSynchronousContinuations = false
        });

        using var subscriber = _subscriberFactory.GetSubscriber();
        // 事件触发时直接写入Channel，无延迟
        subscriber.Update += async (sender, args) =>
        {
            try
            {
                await channel.Writer.WriteAsync(new StockUpdate
                {
                    Symbol = args.Symbol,
                    Price = args.Price
                }, cancellationToken);
            }
            catch (OperationCanceledException)
            {
                // 订阅结束后忽略写入异常即可
            }
        };

        _logger.LogInformation("Subscription started.");

        // 取消信号触发时关闭Channel写入端
        using var cancelReg = cancellationToken.Register(() => channel.Writer.TryComplete());

        // 异步迭代Channel输出流，有数据就推送给客户端，无数据时异步等待不阻塞
        await foreach (var update in channel.Reader.ReadAllAsync(cancellationToken))
        {
            yield return update;
        }

        _logger.LogInformation("Subscription finished.");
    }
}

实现逻辑说明 #

• Channel是.NET官方提供的高性能异步生产者消费者组件，天然适配IAsyncEnumerable的异步迭代模式
• 事件触发后调用WriteAsync写入数据，会立即被ReadAllAsync捕获，无需任何轮询逻辑即可实现无延迟推送
• 整个生命周期和原生Contract First的实现完全一致，取消令牌触发后会自动结束流、释放资源

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