一、客户端流式处理概述
- 客户端流式处理方法在该方法没有接收消息的情况下启动。
requestStream参数用于从客户端读取消息。 返回响应消息时，客户端流式处理调用完成。 - 客户端可以发送多个消息流到服务端，当所有客户端消息流发送结束，调用请求流完结方法，则标记客户端流消息推送结束，等待服务端执行完成。
- 等同于客户端发送批量消息，服务端统一处理。
二、案例介绍
- 接下来 会提供三个案例，用于大家理解
- 第一个客户端流的基础用法
- 第二个客户端流的优化版本
- 第三个客户端的文件流式传输
三、服务端配置（注意：grpc相关配置参考我之前的文章）
// 1.提供公共的实体proto文件
// 2.服务引用对应的proto文件
// 3.定义三个客户流方法
// 公共messages.proto文件
syntax = "proto3";
option csharp_namespace = "GrpcProject";
package grpc.serviceing;
// 请求体
message ServerRequest{
string name = 1;
double height = 2;
int32 age = 3;
bool flag = 4;
float x = 5;
float y = 6;
float z= 7;
repeated string departments = 8;
}
message ServerFileRequest{
bytes fileBytes = 1;
}
// 响应体
message ServerResponse{
bool result = 1;
}
// clientstream.proto 定义service方法
syntax = "proto3";
import "google/protobuf/empty.proto";
import "Protos/messages.proto";
option csharp_namespace = "GrpcProject";
package grpc.serviceing;
service ClientStreamRpc{
// 基础客户端流处理
rpc StreamingFromClient (stream ServerRequest) returns (ServerResponse);
// foreach 客户端流式处理 前提使用C#8 或者更高版本
rpc StreamingClientForeach(stream ServerRequest) returns (ServerResponse);
// 文件流传输
rpc FileStreamFromClient(stream ServerFileRequest) returns (ServerResponse);
}
服务接口实现：
/// <summary>
/// 客户端流式处理
/// </summary>
public class ClientStreamService : ClientStreamRpc.ClientStreamRpcBase
{
/// <summary>
/// 基础访问流模式
/// </summary>
/// <param name="requestStream"></param>
/// <param name="context"></param>
/// <returns></returns>
public override async Task<ServerResponse> StreamingFromClient(IAsyncStreamReader<ServerRequest> requestStream, ServerCallContext context)
{
while (await requestStream.MoveNext())
{
await Console.Out.WriteLineAsync("\r\n-------------------------激光射线------------------------------\r\n");
ServerRequest request = requestStream.Current;
await Handle(request);
}
return new ServerResponse();
}
/// <summary>
/// foreach访问流模式
/// </summary>
/// <param name="requestStream"></param>
/// <param name="context"></param>
/// <returns></returns>
public override async Task<ServerResponse> StreamingClientForeach(IAsyncStreamReader<ServerRequest> requestStream, ServerCallContext context)
{
await foreach (var request in requestStream.ReadAllAsync())
{
await Console.Out.WriteLineAsync("\r\n-------------------------激光射线------------------------------\r\n");
await Handle(request);
}
return new ServerResponse();
}
/// <summary>
/// 读取文件流 组合成完整的文件。
/// </summary>
/// <param name="requestStream"></param>
/// <param name="context"></param>
/// <returns></returns>
public override async Task<ServerResponse> FileStreamFromClient(IAsyncStreamReader<ServerFileRequest> requestStream, ServerCallContext context)
{
ServerResponse serverResponse = new ServerResponse();
serverResponse.Result = false;
// 存储流
MemoryStream ms = new();
await foreach (var request in requestStream.ReadAllAsync())
{
ms.Write(request.FileBytes.Span);
await Console.Out.WriteLineAsync($"记录字节大小：{ms.Length} bytes");
}
string filePath = Path.Combine(Directory.GetCurrentDirectory(), "log.txt");
using FileStream fileStream = new FileStream(filePath, FileMode.OpenOrCreate, FileAccess.ReadWrite);
fileStream.Position = 0;
await fileStream.WriteAsync(ms.ToArray());
fileStream.Flush();
fileStream.Close();
serverResponse.Result = true;
return serverResponse;
}
private async Task Handle(ServerRequest request)
{
if (request != null)
{
foreach (var prop in request.GetType().GetProperties())
{
if (prop.CanRead && prop.GetValue(request) is not null)
{
await Console.Out.WriteLineAsync($"property name：{prop.Name}；value：{prop.GetValue(request)}");
}
}
}
}
}await foreach (var request in requestStream.ReadAllAsync()) 版本是C#8及以上版本才支持，这个需要注意！
四、客户端配置
- 引用proto文件，配置为客户端类型
- 根据编译生成的函数进行传参调用
public partial class ClientStreamForm : Form
{
private readonly string _url;
public ClientStreamForm(IConfiguration configuration)
{
InitializeComponent();
_url = configuration.GetConnectionString("ConnectionString");
}
private async void button1_Click(object sender, EventArgs e)
{
var channel = GrpcChannel.ForAddress(_url);
var client = new ClientStreamRpc.ClientStreamRpcClient(channel);
var streamingFromClient = client.StreamingFromClient();
for (int i = 0; i < 3; i++)
{
ServerRequest request = new();
request.Age = i;
request.X = new Random(30).Next(50);
request.Y = new Random(60).Next(50);
request.Z = new Random(90).Next(50);
request.Flag = i % 2 == 0 ? true : false;
request.Height = 12;
request.Departments.Add($"{i}");
await streamingFromClient.RequestStream.WriteAsync(request);
}
await streamingFromClient.RequestStream.CompleteAsync();
}
private async void btnGrpcOptimize_Click(object sender, EventArgs e)
{
var channel = GrpcChannel.ForAddress(_url);
var client = new ClientStreamRpc.ClientStreamRpcClient(channel);
var streamingFromClient = client.StreamingClientForeach();
var departments = new List<string>()
{
"one",
"two",
"three",
"four",
"five",
"six"
};
ServerRequest request = new();
foreach (var department in departments)
{
request.Age = new Random(20).Next(100);
request.X = new Random(30).Next(50);
request.Y = new Random(60).Next(50);
request.Z = new Random(90).Next(50);
request.Flag = false;
request.Height = 12;
request.Departments.Add(department);
await streamingFromClient.RequestStream.WriteAsync(request);
}
await streamingFromClient.RequestStream.CompleteAsync();
}
private async void btnFile_Click(object sender, EventArgs e)
{
var channel = GrpcChannel.ForAddress(_url);
var client = new ClientStreamRpc.ClientStreamRpcClient(channel);
var streamingFromClient = client.FileStreamFromClient();
var strMessage = "伟大抗日战争的一周年纪念，七月七日，快要到了。全民族的力量团结起来，坚持抗战，坚持统一战线，同敌人作英勇的战争，快一年了。这个战争，在东方历史上是空前的，在世界历史上也将是伟大的，全世界人民都关心这个战争。身受战争灾难、为着自己民族的生存而奋斗的每一个中国人，无日不在渴望战争的胜利。然而战争的过程究竟会要怎么样？能胜利还是不能胜利？能速胜还是不能速胜？很多人都说持久战，但是为什么是持久战？怎样进行持久战？很多人都说最后胜利，但是为什么会有最后胜利？怎样争取最后胜利？这些问题，不是每个人都解决了的，甚至是大多数人至今没有解决的。于是失败主义的亡国论者跑出来向人们说：中国会亡，最后胜利不是中国的。某些性急的朋友们也跑出来向人们说：中国很快就能战胜，无需乎费大气力。这些议论究竟对不对呢？我们一向都说：这些议论是不对的。可是我们说的，还没有为大多数人所了解。一半因为我们的宣传解释工作还不够，一半也因为客观事变的发展还没有完全暴露其固有的性质，还没有将其面貌鲜明地摆在人们之前，使人们无从看出其整个的趋势和前途，因而无从决定自己的整套的方针和做法。现在好了，抗战十个月的经验，尽够击破毫无根据的亡国论，也尽够说服急性朋友们的速胜论了。在这种情形下，很多人要求做个总结性的解释。尤其是对持久战，有亡国论和速胜论的反对意见，也有空洞无物的了解。“卢沟桥事变以来，四万万人一齐努力，最后胜利是中国的。”这样一种公式，在广大的人们中流行着。这个公式是对的，但有加以充实的必要。抗日战争和统一战线之所以能够坚持，是由于许多的因素：全国党派，从共产党到国民党；全国人民，从工人农民到资产阶级；全国军队，从主力军到游击队；国际方面，从社会主义国家到各国爱好正义的人民；敌国方面，从某些国内反战的人民到前线反战的兵士。总而言之，所有这些因素，在我们的抗战中都尽了他们各种程度的努力。每一个有良心的人，都应向他们表示敬意。我们共产党人，同其他抗战党派和全国人民一道，唯一的方向，是努力团结一切力量，战胜万恶的日寇。今年七月一日，是中国共产党建立的十七周年纪念日。为了使每个共产党员在抗日战争中能够尽其更好和更大的努力，也有着重地研究持久战的必要。因此，我的讲演就来研究持久战。和持久战这个题目有关的问题，我都准备说到；但是不能一切都说到，因为一切的东西，不是在一个讲演中完全说得了的。";
var spanLength = strMessage.Length / 10;
for (int i = 0; i < 10; i++)
{
var startIndex = spanLength * i;
string spanMessage;
if (i == 9)
{
spanMessage = strMessage.Substring(startIndex);
}
else
{
spanMessage = strMessage.Substring(startIndex, spanLength - 1);
}
var strBytes = Encoding.UTF8.GetBytes(spanMessage);
var byteString = ByteString.CopyFrom(strBytes, 0, strBytes.Length);
await streamingFromClient.RequestStream.WriteAsync(new ServerFileRequest() { FileBytes = byteString });
Task.Delay(1000).Wait();
}
await streamingFromClient.RequestStream.CompleteAsync();
}
}- IConfiguration 接口在program类中进行注入，读取appsettings.json文件
- 调用接口查看执行结果
- 客户端流基础模式
2.客户端流foreach模式
3.文件流处理模式
- 客户端流基础模式
五、源码地址
链接：https://pan.baidu.com/s/1PnLhysfGbVxC1ecpu7XReA
提取码：l6w0
