ABP VNext + gRPC 双向流：实时数据推送与订阅场景实现

🚀 ABP VNext + gRPC 双向流：实时数据推送与订阅场景实现

---

---

📄 背景与动机

gRPC 的三种流式调用（Server Streaming、Client Streaming、Bidirectional Streaming）是现代实时系统的关键利器。尤其在 ABP VNext 架构中，Bidirectional Streaming 能支持即时、稳定、可控的双工通信，广泛应用于日志推送、聊天系统、行情订阅等场景。

---

🧰 环境与依赖

• .NET SDK：6.0+

• ABP VNext：6.x+

• NuGet 包：

  • Grpc.AspNetCore
  • Volo.Abp.Grpc

• appsettings.json 示例（开发环境 HTTP/2）：

  {"Kestrel": {"Endpoints": {"Grpc": {"Url": "http://localhost:5001","Protocols": "Http2"}}}
  }
  

  生产环境 强烈建议使用 HTTPS/TLS，并在 Kestrel 中配置证书，避免明文传输。

---

🔁 流式模型对比

---

🔄 双向流时序图

---

🚀 ABP 集成 gRPC

// Program.cs 或 Module.ConfigureServices
services.AddSingleton<AuthInterceptor>();
services.AddGrpc(options =>
{options.EnableDetailedErrors = true;                              // 详细错误// 通过工厂方式注入拦截器，确保 IServiceProvider 可用options.Interceptors.Add(sp => sp.GetRequiredService<AuthInterceptor>());// 注册 gzip 压缩（需 AddCompressionProviders）options.ResponseCompressionAlgorithm = "gzip";
})
.AddCompressionProviders();Configure<AbpGrpcOptions>(opt =>
{opt.ProtoRootFolder = "Protos";
});// 中间件管道
app.UseRouting();
app.UseAuthentication();
app.UseAuthorization();
app.UseGrpcEndpoints();

---

📄 Proto 文件

syntax = "proto3";
package realtime;service ChatService {// 双向流：客户端和服务端同时读写rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}message ChatMessage {string user      = 1;string message   = 2;int64  timestamp = 3;
}

---

🎙️ 服务端实现：ChatService

using System.Threading.Channels;
using Grpc.Core;
using Microsoft.Extensions.Logging;public class ChatService : ChatServiceBase
{private readonly ILogger<ChatService> _logger;public ChatService(ILogger<ChatService> logger){_logger = logger;}public override async Task Chat(IAsyncStreamReader<ChatMessage> requestStream,IServerStreamWriter<ChatMessage> responseStream,ServerCallContext context){var ct = context.CancellationToken;// 使用 Channel 实现反压示例var channel = Channel.CreateBounded<ChatMessage>(new BoundedChannelOptions(1000){FullMode = BoundedChannelFullMode.Wait});// 读取客户端写入到 Channelvar readerTask = Task.Run(async () =>{await foreach (var msg in requestStream.ReadAllAsync(ct)){await channel.Writer.WriteAsync(msg, ct);}channel.Writer.Complete();}, ct);// 从 Channel 写到 responseStreamvar writerTask = Task.Run(async () =>{await foreach (var msg in channel.Reader.ReadAllAsync(ct)){var time = DateTimeOffset.FromUnixTimeSeconds(msg.Timestamp).ToLocalTime().ToString("o"); // ISO8601_logger.LogInformation("[{Time}] {User}: {Message}", time, msg.User, msg.Message);try{await responseStream.WriteAsync(msg, ct);}catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled){_logger.LogWarning("Client cancelled the stream");break;}catch (RpcException ex) when (ex.StatusCode == StatusCode.DeadlineExceeded){_logger.LogWarning("Write timed out");break;}catch (Exception ex){_logger.LogError(ex, "Failed to write message");break;}}}, ct);await Task.WhenAll(readerTask, writerTask);}
}

---

🖥️ 客户端实现（Console 示例）

using Grpc.Net.Client;
using Grpc.Core;using var channel = GrpcChannel.ForAddress("https://localhost:5001",           // 生产请用 httpsnew GrpcChannelOptions{MaxReceiveMessageSize = 4 * 1024 * 1024,MaxSendMessageSize    = 4 * 1024 * 1024,});
var client = new ChatService.ChatServiceClient(channel);
using var call = client.Chat();var send = Task.Run(async () =>
{while (true){var line = Console.ReadLine();if (string.IsNullOrWhiteSpace(line)){await call.RequestStream.CompleteAsync();  // 优雅结束上传return;                                    // 结束 send 任务}await call.RequestStream.WriteAsync(new ChatMessage{User      = "User1",Message   = line,Timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds()});}
});var receive = Task.Run(async () =>
{try{await foreach (var reply in call.ResponseStream.ReadAllAsync()){Console.WriteLine($"{reply.User}: {reply.Message}");}}catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled){Console.WriteLine("Stream closed by server");}catch (Exception ex){Console.WriteLine($"Error receiving: {ex.Message}");}
});await Task.WhenAll(send, receive);

---

🔐 安全与拦截器

using Grpc.Core;
using Microsoft.Extensions.Logging;public class AuthInterceptor : Interceptor
{private bool ValidateToken(string token) => /* your logic */ true;private async Task<T> HandleAuth<T>(ServerCallContext context, Func<Task<T>> next){var header = context.RequestHeaders.FirstOrDefault(h => h.Key.Equals("authorization", StringComparison.OrdinalIgnoreCase))?.Value;var token = header?.Replace("Bearer ", "");if (string.IsNullOrEmpty(token) || !ValidateToken(token)){throw new RpcException(new Status(StatusCode.Unauthenticated, "Invalid token"));}return await next();}public override Task<TResponse> UnaryServerHandler<TRequest, TResponse>(TRequest request, ServerCallContext context,UnaryServerMethod<TRequest, TResponse> continuation)=> HandleAuth(context, () => base.UnaryServerHandler(request, context, continuation));public override Task<TResponse> ServerStreamingServerHandler<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream,ServerCallContext context, ServerStreamingServerMethod<TRequest, TResponse> continuation)=> HandleAuth(context, () => base.ServerStreamingServerHandler(request, responseStream, context, continuation));public override Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream,ServerCallContext context,ClientStreamingServerMethod<TRequest, TResponse> continuation)=> HandleAuth(context, () => base.ClientStreamingServerHandler(requestStream, context, continuation));public override Task DuplexStreamingServerHandler<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream,IServerStreamWriter<TResponse> responseStream,ServerCallContext context,DuplexStreamingServerMethod<TRequest, TResponse> continuation)=> HandleAuth(context, () => base.DuplexStreamingServerHandler(requestStream, responseStream, context, continuation));
}

---

📊 性能优化与可观测性

• 消息大小限制 & 压缩

  services.AddGrpc(options =>
  {options.MaxReceiveMessageSize       = 4 * 1024 * 1024;options.MaxSendMessageSize          = 4 * 1024 * 1024;options.ResponseCompressionAlgorithm = "gzip";
  });
  

• 反压策略：Channel 限流示例已于服务端实现展示

• 链路追踪：集成 OpenTelemetry，使用 ActivitySource 捕获分布式 TraceId

• 自定义 Metrics：用 Prometheus .NET 客户端在读写异常、超时处打点 Counter/Gauge

---

🧩 模块化与 CI 自动生成 Proto

[DependsOn(typeof(AbpGrpcModule))]
public class ChatGrpcModule : AbpModule { /* ... */ }

<!-- 在 ChatServer.csproj 中自动生成 C# 类 -->
<ItemGroup><Protobuf Include="Protos\chat.proto" GrpcServices="Server" />
</ItemGroup>

---

📂 参考资源

• ABP VNext 官方文档
• gRPC 官方文档
• ASP.NET Core gRPC 教程