可以做推广的门户网站,怎样策划一个营销型网站,wordpress 5.0网易云音乐,今天最新生猪价格gRPC 使用 Protocol buffers 作为接口定义语言#xff08;IDL#xff09;来描述服务接口和输入输出消息的结构#xff0c;目前支持 4 种定义服务方法类型#xff1a;类型说明简单 RPC客户端传入一个请求对象#xff0c;服务端返回一个结果对象客户端流式 RPC客户端传入多个… gRPC 使用 Protocol buffers 作为接口定义语言IDL来描述服务接口和输入输出消息的结构目前支持 4 种定义服务方法类型类型说明简单 RPC客户端传入一个请求对象服务端返回一个结果对象客户端流式 RPC客户端传入多个请求对象服务端返回一个结果对象服务端流式 RPC客户端传入一个请求对象服务端返回多个结果对象双向流式 RPC客户端传入多个请求对象服务端返回多个结果对象RPC 定义简单 RPC一般这种方式使用较多如下定义 SayHello 方法输入 HelloRequest返回 HelloResponse 。1234567891011service HelloService { rpc SayHello (HelloRequest) returns (HelloResponse);}message HelloRequest { string greeting 1;}message HelloResponse { string reply 1;}而流式 RPC 定义与 简单 RPC 的区别只是在请求或返回参数前增加了 stream 关键词如下12345678service HelloService { // 客户端流式 RPC rpc SayHello1 (stream HelloRequest) returns (HelloResponse); // 服务端流式 RPC rpc SayHello2 (HelloRequest) returns (stream HelloResponse); // 双向流式 RPC rpc SayHello3 (stream HelloRequest) returns (stream HelloResponse);}gRPC 能支持流式调用本质是因为 gRPC 通信是基于 HTTP/2 实现的HTTP/2 具有流的概念流是为了实现 HTTP/2 的多路复用。流是服务器和客户端在 HTTP/2 连接内用于交换帧数据的独立双向序列逻辑上可看做一个较为完整的交互处理单元即表达一次完整的资源请求、响应数据交换流程。使用场景在 gRPC 中消息接收大小 GRPC_DEFAULT_MAX_RECV_MESSAGE_LENGTH 默认是 4M如果大于该值则会提示Error: grpc: received message larger than max (xxxxxx vs. 4194304)当然我们可以修改默认值解决问题但如果默认值支持过大对服务器资源也是一种消耗这时候其实应该考虑使用流式调用有效将数据进行分批处理提高性能。示例这里主要介绍一下双向流式 RPC客户端和服务端流式 RPC 类似完整代码请 前往这里查看 。双向流模拟功能是客户端流式输入文件路径服务端针对每个文件每次最多读取 1M 的数据返回客户端拿到数据后生成新文件。接口定义12345678910111213141516171819syntax proto3;package GrpcStream;service StreamTest { // 双向流程 RPC rpc BidirectionalStream(stream BidirectionalStreamRequest) returns (stream BidirectionalStreamResponse) {}}message BidirectionalStreamRequest { // 文件路径 string file_path 1;}message BidirectionalStreamResponse { // 文件路径 string file_path 1; // 数据 bytes data 2;}代码实现这里是基于 .NET Core 3.0 使用 gRPC可以通过 VS 预置的 gRPC 服务 模板来创建服务端创建后将默认的 porto 文件替换成上面的内容。服务端代码实现1234567891011121314151617181920212223242526public override async Task BidirectionalStream(IAsyncStreamReaderBidirectionalStreamRequest requestStream, IServerStreamWriterBidirectionalStreamResponse responseStream, ServerCallContext context){ var i 0; // 监听客户端数据输入 while (await requestStream.MoveNext()) { // 打印次数 Console.WriteLine(i); using var fs File.Open(requestStream.Current.FilePath, FileMode.Open); var leftSize fs.Length; // 1M var buff new byte[1048576]; while (leftSize 0) { var len await fs.ReadAsync(buff); leftSize - len; Console.WriteLine($response {requestStream.Current.FilePath} {len} bytes); // 流式返回数据 await responseStream.WriteAsync(new BidirectionalStreamResponse { FilePath requestStream.Current.FilePath, Data ByteString.CopyFrom(buff, 0, len) }); } }}客户端代码实现1234567891011121314151617181920212223242526272829303132333435363738394041424344454647// 测试文件key 是已存在的文件value 是需要生成的文件static readonly Dictionarystring, string fileDic new Dictionarystring, string(){ {d:\dapr\daprd_windows_amd64.zip, d:\dapr\daprd_windows_amd64_new.zip }, {d:\dapr\injector_windows_amd64.zip, d:\dapr\injector_windows_amd64_new.zip },};static StreamTest.StreamTestClient client;static async Task Main(string[] args){ // 连接 gRPC 服务 var channel GrpcChannel.ForAddress(https://localhost:5001); client new StreamTest.StreamTestClient(channel); await BidirectionalStreamTestAsync(); Console.ReadKey();}static async Task BidirectionalStreamTestAsync(){ using var call client.BidirectionalStream(); var responseTask Task.Run(async () { // 接收返回值 var iterator call.ResponseStream; // 监听服务端数据返回 while (await iterator.MoveNext()) { Console.WriteLine($write to new file {fileDic[iterator.Current.FilePath]} {iterator.Current.Data.Length} bytes); // 写入新文件 using var fs new FileStream(fileDic[iterator.Current.FilePath], FileMode.Append); iterator.Current.Data.WriteTo(fs); } }); var rand new Random(); foreach (var item in fileDic) { // 流式输入 await call.RequestStream.WriteAsync(new BidirectionalStreamRequest { FilePath item.Key }); await Task.Delay(rand.Next(200)); } await call.RequestStream.CompleteAsync(); await responseTask;}执行结果参考资料gRPC Concepts
