在实时数据类应用场景中，比如即时通讯、实时数据监控、多端状态同步等，系统往往需要同时处理大量客户端的并发请求，并且要求数据传输延迟低、吞吐量大。传统的HTTP请求响应模式每次交互都需要建立连接、等待响应，很难满足这类场景的需求。gRPC的双向流模式允许客户端和服务端在单个连接上同时发送多个消息，不需要等待对方响应，非常适合处理高并发实时数据场景，结合c#的高效开发能力，可以快速搭建稳定的实时数据服务。
gRPC双向流核心原理
gRPC基于HTTP/2协议实现，支持四种通信模式，其中双向流模式是客户端和服务端都可以在建立连接后，主动向对方发送多个消息，消息的发送和接收是异步的，不需要遵循请求-响应的顺序。这种模式可以减少连接建立的开销，单个连接就可以承载多个并发的数据流，非常适合高并发场景下的实时数据传输。
在双向流中，客户端调用服务方法后会得到一个流对象，既可以往流里写入消息，也可以从流里读取服务端返回的消息，两边的操作互不阻塞。服务端在接收到客户端的连接请求后，也会拿到对应的流对象，同样可以双向读写消息，实现全双工的实时通信。
c#中搭建gRPC双向流服务
1. 定义proto协议文件
首先需要定义proto文件，声明双向流的服务方法，这里以实时数据上报和推送场景为例，定义数据上报和实时通知的双向流接口。
syntax = "proto3";
option csharp_namespace = "GrpcRealTimeDemo";
package realtime;
// 定义实时数据消息
message RealTimeData {
string client_id = 1;
string data_content = 2;
int64 timestamp = 3;
}
// 定义服务端推送的消息
message ServerPushMessage {
string target_client_id = 1;
string push_content = 2;
}
// 定义双向流服务
service RealTimeStreamService {
// 双向流方法，客户端和服务端都可以发送多个消息
rpc ProcessRealTimeData (stream RealTimeData) returns (stream ServerPushMessage);
}
2. 实现服务端逻辑
在c#项目中引用Grpc.AspNetCore包，然后实现proto中定义的服务接口，处理双向流的读写操作，同时加入简单的并发控制逻辑。
using Grpc.Core;
using GrpcRealTimeDemo;
namespace GrpcRealTimeDemo.Services;
public class RealTimeStreamServiceImpl : RealTimeStreamService.RealTimeStreamServiceBase
{
// 控制并发处理的信号量，限制同时处理的数据流数量，避免资源耗尽
private static readonly SemaphoreSlim _concurrencyLimiter = new SemaphoreSlim(100, 100);
private readonly ILogger<RealTimeStreamServiceImpl> _logger;
public RealTimeStreamServiceImpl(ILogger<RealTimeStreamServiceImpl> logger)
{
_logger = logger;
}
public override async Task ProcessRealTimeData(
IAsyncStreamReader<RealTimeData> requestStream,
IServerStreamWriter<ServerPushMessage> responseStream,
ServerCallContext context)
{
// 等待获取并发处理许可
await _concurrencyLimiter.WaitAsync(context.CancellationToken);
try
{
// 启动两个异步任务，分别处理客户端发来的数据和向客户端推送数据
var readClientDataTask = ReadClientDataAsync(requestStream, context);
var pushServerDataTask = PushServerDataAsync(responseStream, context);
// 等待两个任务完成，任意一个取消就整体取消
await Task.WhenAny(readClientDataTask, pushServerDataTask);
}
finally
{
// 释放并发许可
_concurrencyLimiter.Release();
}
}
// 读取客户端发送的实时数据
private async Task ReadClientDataAsync(IAsyncStreamReader<RealTimeData> requestStream, ServerCallContext context)
{
while (await requestStream.MoveNext(context.CancellationToken))
{
var clientData = requestStream.Current;
_logger.LogInformation($"收到客户端{clientData.ClientId}的数据：{clientData.DataContent}");
// 这里可以加入数据持久化、转发给其他客户端等业务逻辑
}
}
// 向客户端推送服务端消息
private async Task PushServerDataAsync(IServerStreamWriter<ServerPushMessage> responseStream, ServerCallContext context)
{
// 模拟定期向客户端推送消息，实际场景中可以结合消息队列、事件总线触发推送
while (!context.CancellationToken.IsCancellationRequested)
{
await responseStream.WriteAsync(new ServerPushMessage
{
TargetClientId = "all",
PushContent = $"服务端推送消息，时间：{DateTimeOffset.UtcNow.ToUnixTimeSeconds()}"
});
await Task.Delay(1000, context.CancellationToken);
}
}
}
3. 客户端调用示例
客户端同样需要引用对应的proto生成的代码，然后建立连接，启动双向流的读写操作，实现实时数据的发送和接收。
using Grpc.Net.Client;
using GrpcRealTimeDemo;
namespace GrpcClientDemo;
class Program
{
static async Task Main(string[] args)
{
// 创建gRPC客户端通道，地址根据实际服务端配置调整
using var channel = GrpcChannel.ForAddress("http://127.0.0.1:5001");
var client = new RealTimeStreamService.RealTimeStreamServiceClient(channel);
// 调用双向流方法，获取流对象
using var call = client.ProcessRealTimeData();
// 启动异步任务接收服务端推送的消息
var readServerPushTask = Task.Run(async () =>
{
while (await call.ResponseStream.MoveNext(CancellationToken.None))
{
var serverMessage = call.ResponseStream.Current;
Console.WriteLine($"收到服务端推送：{serverMessage.PushContent}");
}
});
// 模拟客户端持续发送实时数据
var random = new Random();
for (int i = 0; i < 10; i++)
{
await call.RequestStream.WriteAsync(new RealTimeData
{
ClientId = "client_001",
DataContent = $"实时数据_{random.Next(1000)}",
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds()
});
await Task.Delay(500);
}
// 发送完数据后标记请求流结束
await call.RequestStream.CompleteAsync();
// 等待服务端推送任务完成
await readServerPushTask;
}
}
高并发场景优化方案
在处理高并发实时数据时，除了基础的双向流实现，还需要做对应的优化来提升服务的稳定性和吞吐量：
- 使用信号量或者连接数限制，控制同时处理的双向流数量，避免大量连接同时建立导致服务端资源耗尽
- 对于需要转发、持久化的数据，采用异步处理的方式，不要阻塞双向流的读写线程，避免影响实时性
- 结合消息队列解耦数据生产和消费，服务端收到客户端数据后写入消息队列，由消费者异步处理，减少流处理的耗时
- 合理设置HTTP/2的连接参数，比如调整帧大小、流控窗口，提升单个连接的传输效率
- 对客户端连接做心跳检测，及时关闭无效连接，释放资源
注意事项
使用gRPC双向流处理高并发实时数据时，需要注意流的生命周期管理，避免流没有正确关闭导致内存泄漏。同时要做好异常处理，当客户端断开连接时，服务端的流操作要能正确捕获取消信号，释放相关资源。另外，双向流适合持续的数据交互场景，如果是单次请求的场景，使用普通的请求响应模式会更合适。