.net core 用grpc实现微服务

时间：2016-11-04 20:10:39 阅读：424 评论：0 收藏：0 [点我收藏+]

标签：manual each tty sea 发布 size list plugin .json

GRPC?是Google发布的一个开源、高性能、通用RPC（Remote Procedure Call）框架。提供跨语言、跨平台支持。以下以.NET Core 使用控制台、docker中演示如何使用GRPC框架。

? ?

软件版本

.net core :1.0.1

GRPC:1.0.1-pre1

? ?

1.定义服务

使用proto3语法定义一个服务，主要测试package、import、常用类型的测试,

proto3语法: https://developers.google.com/protocol-buffers/docs/proto3

? ?

定义一个result.proto

? ?

syntax = "proto3";

package App.RPC.Model;

message Response{

bool sucess=1;

string message=2;

}

? ?

定义RPCDemoService.proto文件如下：

syntax = "proto3";

package App.RPC.Service;

import "result.proto";

? ?

service RPCDemoService{

rpc Add(DemoRequest) returns (App.RPC.Model.Response){}

rpc GetById(DemoId) returns (DemoRequest){}

rpc Get(Search) returns (DemoList){}

}

message DemoId{

int32 Id=1;

}

? ?

message Search{

int32 page=1;

int32 size=2;

string query=3;

}

? ?

message DemoRequest{

string Id=1;

int32 CommentId=2;

bool IsDeleted=3;

}

? ?

message DemoList{

repeated DemoRequest details = 1;

}

? ?

2.将服务生成类文件:

项目引用nuget包Grpc.Tools 1.0.0 或通过命令安装这个程序包,然后找到文件路径

先配置packages\Grpc.Tools\1.0.0\tools\windows_x64\protoc.exe环境变量

protoc.exe --csharp_out=d:\grpcdemo\code\ --grpc_out=d:\grpcdemo\code\ --plugin=protoc-gen-grpc=yourpath\.nuget\packages\Grpc.Tools\1.0.0\tools\windows_x64\grpc_csharp_plugin.exe result.proto

protoc.exe --csharp_out=d:\grpcdemo\code\ --grpc_out=d:\grpcdemo\code\ --plugin=protoc-gen-grpc=yourpath\.nuget\packages\Grpc.Tools\1.0.0\tools\windows_x64\grpc_csharp_plugin.exe RPCDemoService.proto

? ?

3.创建.net core 类型项目App.RPCDemo

Project.json文件内容如下：

{

"version": "1.0.0-*",

"dependencies": {

"Grpc.Tools": "1.0.0"

"frameworks": {

"netstandard1.6": {

"imports": "dnxcore50",

"dependencies": {

"NETStandard.Library": "1.6.0",

"Grpc": "1.0.1-pre1",

"Grpc.Core": "1.0.1-pre1",

"Google.Protobuf": "3.1.0",

"System.Interactive.Async": "3.0.0"

}

? ?

4.创建服务端App.RPCDemoServer

因为要在docker 中进行测试，官方网站并没有docker 下的例子，也没有说这个rpc在生产环境中如何hosting; 通过查看.net core的host源码在Microsoft.AspNetCore.Hosting.Internal.ApplicationLifetime这个类文件有实现，把这个文件的内容直接copy过来

类的部分内容如下：

private readonly CancellationTokenSource _startedSource = new CancellationTokenSource();

private readonly CancellationTokenSource _stoppingSource = new CancellationTokenSource();

private readonly CancellationTokenSource _stoppedSource = new CancellationTokenSource();

? ?

/// <summary>

/// Triggered when the application host has fully started and is about to wait

/// for a graceful shutdown.

/// </summary>

public CancellationToken ApplicationStarted => _startedSource.Token;

? ?

/// <summary>

/// Triggered when the application host is performing a graceful shutdown.

/// Request may still be in flight. Shutdown will block until this event completes.

/// </summary>

public CancellationToken ApplicationStopping => _stoppingSource.Token;

? ?

/// <summary>

/// Triggered when the application host is performing a graceful shutdown.

/// All requests should be complete at this point. Shutdown will block

/// until this event completes.

/// </summary>

public CancellationToken ApplicationStopped => _stoppedSource.Token;

? ?

google Grpc中的server类型没有接口，定义一个接口Iserver

namespace App.RPC.Core

{

public interface IServer

{

void Start();

? ?

void Stop();

? ?

Status State { get; set; }

}

? ?

public enum Status

{

None,

Stop,

Running

? ?

}

? ?

在创建一个hosting 文件内容如下：

? ?

public static class RpcHostExtensions

{

public static void Run(this IServer server)

{

var done = new ManualResetEventSlim(false);

using (var cts = new CancellationTokenSource())

{

Action shutdown = () =>

{

if (!cts.IsCancellationRequested)

{

server.Stop();

Console.WriteLine("Rpc Service is shutting down...");

cts.Cancel();

}

done.Wait();

};

? ?

#if NETSTANDARD1_5

var assemblyLoadContext = AssemblyLoadContext.GetLoadContext(typeof(WebHostExtensions).GetTypeInfo().Assembly);

assemblyLoadContext.Unloading += context => shutdown();

#endif

Console.CancelKeyPress += (sender, eventArgs) =>

{

shutdown();

// Don‘t terminate the process immediately, wait for the Main thread to exit gracefully.

eventArgs.Cancel = true;

};

? ?

server.Run(cts.Token, "Rpc Service started. Press Ctrl+C to shut down.");

done.Set();

}

? ?

/// <summary>

/// Runs a web application and block the calling thread until token is triggered or shutdown is triggered.

/// </summary>

/// <param name="host">The <see cref="IWebHost"/> to run.</param>

/// <param name="token">The token to trigger shutdown.</param>

public static void Run(this IServer server, CancellationToken token)

{

server.Run(token, shutdownMessage: null);

}

? ?

private static void Run(this IServer server, CancellationToken token, string shutdownMessage)

{

if (server.State != Status.Running)

{

server.Start();

}

? ?

var applicationLifetime = new ApplicationLifetime();

if (!string.IsNullOrEmpty(shutdownMessage))

{

Console.WriteLine(shutdownMessage);

}

? ?

token.Register(state =>

{

((ApplicationLifetime)state).StopApplication();

applicationLifetime);

? ?

applicationLifetime.ApplicationStopping.WaitHandle.WaitOne();

? ?

}

? ?

实现服务RPCDemoImpl

public class RPCDemoImpl : RPCDemoService.RPCDemoServiceBase

{

public override Task<Response> Add(DemoRequest request, ServerCallContext context)

{

return Task.FromResult(new Response { Message = "成功" + context.Host + DateTime.Now.Ticks.ToString(), Sucess = true });

}

? ?

public override Task<DemoList> Get(Search request, ServerCallContext context)

{

var result = new DemoList();

result.Details.Add(new DemoRequest()

{

CommentId = 1,

Id = DateTime.Now.Ticks.ToString(),

IsDeleted = false

});

return Task.FromResult(result);

? ?

}

? ?

public override Task<DemoRequest> GetById(DemoId request, ServerCallContext context)

{

return Task.FromResult(new DemoRequest()

{

CommentId = request.Id,

Id = DateTime.Now.Ticks.ToString(),

IsDeleted = false

});

}

? ?

program文件如下：

public class Program

{

public static void Main(string[] args)

{

string host = "0.0.0.0";

int port = 9007;

var dic = Common.GetArgs(args);

if (dic != null && dic.Count > 0)

{

string tempHost;

string tempPort;

if (dic.TryGetValue("host", out tempHost))

{

host = tempHost;

}

if (dic.TryGetValue("port", out tempPort))

{

port = Convert.ToInt32(tempPort);

}

? ?

GrpcServer server = new GrpcServer

{

Services = { RPCDemoService.BindService(new RPCDemoImpl()) },

Ports = { new ServerPort(host, port, ServerCredentials.Insecure) }

};

Console.WriteLine("Google Grpc Starting");

foreach (var item in server.Ports)

{

Console.WriteLine(string.Format("RPC server {0} listening on port {1}", item.Host, item.Port));

}

server.Run();

? ?

}

? ?

编译发布后运行如下：

技术分享

? ?

客户端程序

public static void Main(string[] args)

{

string host = "127.0.0.1";

string port = "9007";

long length = 10;

var dic = Common.GetArgs(args);

if (dic != null && dic.Count > 0)

{

string tempHost;

string tempPort, tempLength;

? ?

if (dic.TryGetValue("host", out tempHost))

{

host = tempHost;

}

if (dic.TryGetValue("port", out tempPort))

{

port = tempPort;

}

? ?

if (dic.TryGetValue("repeat", out tempLength))

{

length = Convert.ToInt64(tempLength);

}

? ?

Channel channel = new Channel(string.Format("{0}:{1}", host, port), ChannelCredentials.Insecure);

var client = new RPCDemoService.RPCDemoServiceClient(channel);

? ?

var stopwatch = Stopwatch.StartNew();

for (var i = 0; i < length; i++)

{

? ?

var reply = client.GetById(new DemoId() { Id = i });

Console.WriteLine("receive" + JsonConvert.SerializeObject(reply));

}

stopwatch.Stop();

? ?

Console.WriteLine(string.Format("repeat={0}, time={1} Milliseconds, time/repeat={2}", length, stopwatch.ElapsedMilliseconds, stopwatch.ElapsedMilliseconds / (float)length));

Console.ReadKey();

? ?

channel.ShutdownAsync().Wait();

? ?

}

? ?

编译发布运行如下：

? ?

技术分享