C# 之 Grpc

1. Grpc 是什么？

Google RPC A high-performance, open source universal RPC framework

官方地址：https://www.grpc.io/

Grpc 开始是由google 开发的，是一款语言中立、平台中立、开源的远程调用（RPC）系统.

执行流程：

01.Proto文件

接口—定义不同语言的实现规范 需要个工具—转换C#代码(webservice代理) 定义了协议接口和数据格式 不同语言—相同文件---等于接口

类型映射：

参开地址：https://developers.google.cn/protocol-buffers/docs/proto3#scalar

序列化支持 PB（Protocol buffer）和 JSON， PB 是一种语言无关的高性能序列化框架，基于 HTTP/2 + PB, 保障了 RPC 调用的高性能。

 02.创建Grpc工程。

gRPC支持4种流

1. 简单 RPC（Unary RPC）
2. 服务端流式 RPC （Server streaming RPC）
3. 客户端流式 RPC （Client streaming RPC）
4. 双向流式 RPC（Bi-directional streaming RPC）

 

 1.简单RPC

syntax = "proto3";

option csharp_namespace = "GrpcDemoClients";

package mathUtilsPackage;

// The greeting service definition.
service MathUtils {
  // Sends a greeting
  rpc Add (AddRequest) returns (AddReply);
}

// The request message containing the user's name.
message AddRequest {
  int32 left = 1;
  int32 right = 2;
}

// The response message containing the greetings.
message AddReply {
  int32 total = 1;
}

using System.Threading.Tasks;
using Grpc.Core;

namespace GrpcDemoClients.Services
{
    public class MathUtilsService:MathUtils.MathUtilsBase
    {
        public override Task<AddReply> Add(AddRequest request, ServerCallContext context)
        {
            return Task.FromResult(new AddReply
            {
                Total = request.Left + request.Right
            });
        }
    }
}

配置对外公开的接口。

服2.务端配置：

 客户端配置：

 客户端调用：

   private static async Task TestMathTotal()
        {
            using var channel = GrpcChannel.ForAddress(" https://localhost:5001");
            var client = new MathUtils.MathUtilsClient(channel);
            var reply = await client.AddAsync(new AddRequest
            {
                Left=1,
                Right=2
            });
            Console.WriteLine("Return the value is:" + reply.Total);

        }

2.服务端流式 RPC  一次请求，多次返回。

proto 文件：

syntax = "proto3";

option csharp_namespace = "GrpcDemoClients";

package serverRpcUtilsPackage;

// The greeting service definition.
service ServerRpcUtils {
  // Sends a greeting
    rpc SelfIncreaseServer(IntArrayModel) returns (stream BathTheCatResp);
}

message IntArrayModel{
    repeated int32 Numbers=1;//集合
}
message BathTheCatResp{
    string Message=1;
}

 server 实现：

using System;
using System.Threading.Tasks;
using Grpc.Core;

namespace GrpcDemoClients.Services
{
    public class ServerRpcService: ServerRpcUtils.ServerRpcUtilsBase
    {
        public async override Task SelfIncreaseServer(IntArrayModel request, IServerStreamWriter<BathTheCatResp> responseStream, ServerCallContext context)
        {
            foreach (var item in request.Numbers)
            {
                int number = item;
                Console.WriteLine($"This is {number} invoke");
                await responseStream.WriteAsync(new BathTheCatResp() { Message = $"number++ ={++number}！" });
                await Task.Delay(500);//此处主要是为了方便客户端能看出流调用的效果
            }
        }
    }
}

调用：

 private static async Task TestServerRpc()
        {
            using var channel = GrpcChannel.ForAddress(" https://localhost:5001");
            var client = new ServerRpcUtils.ServerRpcUtilsClient(channel);

            var resp = new IntArrayModel(); 
            for (int i = 0; i < 15; i++)
            {
                resp.Numbers.Add(i);
            }

            var bathCat =  client.SelfIncreaseServer(resp);

            var bathCatRespTask = Task.Run(async () =>
            {
                await foreach (var resp in bathCat.ResponseStream.ReadAllAsync())
                {
                    Console.WriteLine(resp.Message);
                    Console.WriteLine($"This is  Response {Thread.CurrentThread.ManagedThreadId}");
                    Console.WriteLine("**********************************");
                }
            });
            Console.WriteLine("客户端已发送完10个id");
            //开始接收响应
            await bathCatRespTask;

        }

3.客户端流式 RPC   多次请求，一次性返回。

proto 文件：

syntax = "proto3";

option csharp_namespace = "GrpcDemoClients";

package serverRpcUtilsPackage;

// The greeting service definition.
service ServerRpcUtils {
  // Sends a greeting
    rpc SelfIncreaseServer(IntArrayModel) returns (stream BathTheCatResp);
    rpc SelfIncreaseClient(stream BathTheCatReq) returns (IntArrayModel);
}

message IntArrayModel{
    repeated int32 Numbers=1;//集合
}
message BathTheCatResp{
    string Message=1;
}
message BathTheCatReq{
    int32 Id=1;
}

 server 实现：

 public async override Task<IntArrayModel> SelfIncreaseClient(IAsyncStreamReader<BathTheCatReq> requestStream, ServerCallContext context)
        {
            IntArrayModel intArrayModel = new IntArrayModel();
            while (await requestStream.MoveNext())
            {
                intArrayModel.Numbers.Add(requestStream.Current.Id + 1);
                Console.WriteLine($"SelfIncreaseClient Number {requestStream.Current.Id} 获取到并处理.");
                Thread.Sleep(100);
            }
            return intArrayModel;
        }

调用：

 private static async Task TestClientRpc()
        {
            using var channel = GrpcChannel.ForAddress(" https://localhost:5001");
            var client = new ServerRpcUtils.ServerRpcUtilsClient(channel);
            var bathCat = client.SelfIncreaseClient();
            for (int i = 0; i < 10; i++)
            {
                await bathCat.RequestStream.WriteAsync(new BathTheCatReq() { Id = new Random().Next(0, 20) });
                await Task.Delay(100);
                Console.WriteLine($"This is {i} Request {Thread.CurrentThread.ManagedThreadId}");
            }

            Console.WriteLine("**********************************");
            //发送完毕
            await bathCat.RequestStream.CompleteAsync();
            Console.WriteLine("客户端已发送完10个id");
            Console.WriteLine("接收结果：");

            foreach (var item in bathCat.ResponseAsync.Result.Numbers)
            {
                Console.WriteLine($"This is {item} Result");
            }
            Console.WriteLine("**********************************");
        }

4.双向流式 RPC  多次请求，多次返回。

 proto 文件：

rpc SelfIncreaseDouble(stream BathTheCatReq) returns ( stream BathTheCatResp);

 server 实现：

  public async override Task SelfIncreaseDouble(IAsyncStreamReader<BathTheCatReq> requestStream, IServerStreamWriter<BathTheCatResp> responseStream, ServerCallContext context)
        {

            while (await requestStream.MoveNext())
            {
                Console.WriteLine($"SelfIncreaseDouble Number {requestStream.Current.Id} 获取到并处理.");
                await responseStream.WriteAsync(new BathTheCatResp() { Message = $"number++ ={requestStream.Current.Id + 1}！" });
                await Task.Delay(500);//此处主要是为了方便客户端能看出流调用的效果
            }
        }

调用：

 private static async Task TestDoubleRpc()
        {
            using var channel = GrpcChannel.ForAddress(" https://localhost:5001");
            var client = new ServerRpcUtils.ServerRpcUtilsClient(channel);

            var bathCat = client.SelfIncreaseDouble();
            var bathCatRespTask = Task.Run(async () =>
            {
                await foreach (var resp in bathCat.ResponseStream.ReadAllAsync())
                {
                    Console.WriteLine(resp.Message);
                    Console.WriteLine($"This is  Response {Thread.CurrentThread.ManagedThreadId}");
                    Console.WriteLine("**********************************");
                }
            });
            for (int i = 0; i < 10; i++)
            {
                await bathCat.RequestStream.WriteAsync(new BathTheCatReq() { Id = new Random().Next(0, 20) });
                await Task.Delay(100);
                Console.WriteLine($"This is {i} Request {Thread.CurrentThread.ManagedThreadId}");
                Console.WriteLine("**********************************");
            }
            //发送完毕
            await bathCat.RequestStream.CompleteAsync();
            Console.WriteLine("客户端已发送完10个id");
            Console.WriteLine("接收结果：");
            //开始接收响应
            await bathCatRespTask;
        }

补充：

无参数请求

  rpc Count(google.protobuf.Empty) returns (CountResult);//无参数Empty

  public override Task<CountResult> Count(Empty request, ServerCallContext context)
        {

            return Task.FromResult(new CountResult()
            {
                Count = DateTime.Now.Year
            });
        }