深入浅出 gRPC 01:gRPC 服务端创建和调用原理

- 14 mins

1. RPC 入门

1.1 RPC 框架原理

RPC 框架的目标就是让远程服务调用更加简单、透明,RPC 框架负责屏蔽底层的传输方式(TCP 或者 UDP)、序列化方式(XML/Json/ 二进制)和通信细节。服务调用者可以像调用本地接口一样调用远程的服务提供者,而不需要关心底层通信细节和调用过程。

RPC 框架的调用原理图如下所示:

1.2 业界主流的 RPC 框架

业界主流的 RPC 框架整体上分为三类:

随着微服务的发展,基于语言中立性原则构建微服务,逐渐成为一种主流模式,例如对于后端并发处理要求高的微服务,比较适合采用 Go 语言构建,而对于前端的 Web 界面,则更适合 Java 和 JavaScript。
因此,基于多语言的 RPC 框架来构建微服务,是一种比较好的技术选择。例如 Netflix,API 服务编排层和后端的微服务之间就采用 gRPC 进行通信。

1.3 gRPC 简介

gRPC 是一个高性能、开源和通用的 RPC 框架,面向服务端和移动端,基于 HTTP/2 设计。

1.3.1 gRPC 概览

gRPC 是由 Google 开发并开源的一种语言中立的 RPC 框架,当前支持 C、Java 和 Go 语言,其中 C 版本支持 C、C++、Node.js、C# 等。当前 Java 版本最新 Release 版为 1.5.0。

Git 地址如下:grpc-java

gRPC 的调用示例如下所示:

1.3.2 gRPC 特点

2. gRPC 服务端创建

以官方的 helloworld 为例,介绍 gRPC 服务端创建以及 service 调用流程(采用简单 RPC 模式)。

2.1 服务端创建业务代码

服务定义如下(helloworld.proto):

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}
message HelloRequest {
  string name = 1;
}
message HelloReply {
  string message = 1;
}

服务端创建代码如下(HelloWorldServer 类):

private void start() throws IOException {
    /* The port on which the server should run */
    int port = 50051;
    server = ServerBuilder.forPort(port)
        .addService(new GreeterImpl())
        .build()
        .start();
...

其中,服务端接口实现类(GreeterImpl)如下所示:

static class GreeterImpl extends GreeterGrpc.GreeterImplBase {
    @Override
    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
      responseObserver.onNext(reply);
      responseObserver.onCompleted();
    }
  }

2.2 服务端创建流程

gRPC 服务端创建采用 Build 模式,对底层服务绑定、transportServer 和 NettyServer 的创建和实例化做了封装和屏蔽,让服务调用者不用关心 RPC 调用细节,整体上分为三个过程:

下面我们看下 gRPC 服务端创建流程:

gRPC 服务端创建关键流程分析:

2.3 服务端 service 调用流程

gRPC 的客户端请求消息由 Netty Http2ConnectionHandler 接入,由 gRPC 负责将 PB 消息(或者 JSON)反序列化为 POJO 对象,然后通过服务定义查询到该消息对应的接口实例,发起本地 Java 接口调用,调用完成之后,将响应消息反序列化为 PB(或者 JSON),通过 HTTP2 Frame 发送给客户端。
流程并不复杂,但是细节却比较多,整个 service 调用可以划分为如下四个过程:

2.3.1 gRPC 请求消息接入

gRPC 的请求消息由 Netty HTTP/2 协议栈接入,通过 gRPC 注册的 Http2FrameListener,将解码成功之后的 HTTP Header 和 HTTP Body 发送到 gRPC 的 NettyServerHandler 中,实现基于 HTTP/2 的 RPC 请求消息接入。

gRPC 请求消息接入流程如下:

关键流程解读如下:

2.3.2 gRPC 消息头和消息体处理

gRPC 消息头的处理入口是 NettyServerHandler 的 onHeadersRead(),处理流程如下所示:

处理流程如下:

gRPC 消息体的处理入口是 NettyServerHandler 的 onDataRead(),处理流程如下所示:

消息体处理比较简单,下面就关键技术点进行讲解:

2.3.3 内部的服务路由和调用

内部的服务路由和调用,主要包括如下几个步骤:

具体流程如下所示:

中间的交互流程比较复杂,涉及的类较多,但是关键步骤主要有三个:

2.3.4 响应消息发送

响应消息的发送由 StreamObserver 的 onNext 触发,流程如下所示:

响应消息的发送原理如下:

需要指出的是,请求消息的接收、服务调用以及响应消息发送,多次发生 NIO 线程和应用线程之间的互相切换,以及并行处理。因此上述流程中的一些步骤,并不是严格按照图示中的顺序执行的,后续线程模型章节,会做分析和介绍。

3. 源码分析

3.1 主要类和功能交互流程

3.1.1 gRPC 请求消息头处理


gRPC 请求消息头处理涉及的主要类库如下:

3.1.2 gRPC 请求消息体处理和服务调用


3.1.3 gRPC 响应消息处理


需要说明的是,响应消息的发送由调用服务端接口的应用线程执行,在本示例中,由 SerializingExecutor 进行调用。
当请求消息头被封装成 SendResponseHeadersCommand 并被插入到 WriteQueue 之后,后续操作由 Netty 的 NIO 线程 NioEventLoop 负责处理。
应用线程继续发送响应消息体,将其封装成 SendGrpcFrameCommand 并插入到 WriteQueue 队列中,由 Netty 的 NIO 线程 NioEventLoop 处理。响应消息的发送严格按照顺序:即先消息头,后消息体。

3.2 源码分析

了解 gRPC 服务端消息接入和 service 调用流程之后,针对主要的流程和类库,进行源码分析,以加深对 gRPC 服务端工作原理的了解。

3.2.1 Netty 服务端创建

基于 Netty 的 HTTP/2 协议栈,构建 gRPC 服务端,Netty HTTP/2 协议栈初始化代码如下所示(创建 NettyServerHandler,NettyServerHandler 类):

 frameWriter = new WriteMonitoringFrameWriter(frameWriter, keepAliveEnforcer);
    Http2ConnectionEncoder encoder = new DefaultHttp2ConnectionEncoder(connection, frameWriter);
    Http2ConnectionDecoder decoder = new FixedHttp2ConnectionDecoder(connection, encoder,
        frameReader);
    Http2Settings settings = new Http2Settings();
    settings.initialWindowSize(flowControlWindow);
    settings.maxConcurrentStreams(maxStreams);
    settings.maxHeaderListSize(maxHeaderListSize);
    return new NettyServerHandler(
        transportListener, streamTracerFactories, decoder, encoder, settings, maxMessageSize,
        keepAliveTimeInNanos, keepAliveTimeoutInNanos,
        maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,
        keepAliveEnforcer);

创建 gRPC FrameListener,作为 Http2FrameListener,监听 HTTP/2 消息的读取,回调到 NettyServerHandler 中(NettyServerHandler 类):

decoder().frameListener(new FrameListener());

将 NettyServerHandler 添加到 Netty 的 ChannelPipeline 中,接收和发送 HTTP/2 消息(NettyServerTransport 类):

ChannelHandler negotiationHandler = protocolNegotiator.newHandler(grpcHandler);
    channel.pipeline().addLast(negotiationHandler);

gRPC 服务端请求和响应消息统一由 NettyServerHandler 拦截处理,相关方法如下:

NettyServerHandler 是 gRPC 应用侧和底层协议栈的桥接类,负责将原生的 HTTP/2 消息调度到 gRPC 应用侧,同时将应用侧的消息发送到协议栈。

3.2.2 服务实例创建和绑定

gRPC 服务端启动时,需要将调用的接口实现类实例注册到内部的服务注册中心,用于后续的接口调用,关键代码如下(InternalHandlerRegistry 类):

Builder addService(ServerServiceDefinition service) {
      services.put(service.getServiceDescriptor().getName(), service);
      return this;
    }

服务接口绑定时,由 Proto3 工具生成代码,重载 bindService() 方法(GreeterImplBase 类):

@java.lang.Override public final io.grpc.ServerServiceDefinition bindService() {
      return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
          .addMethod(
            METHOD_SAY_HELLO,
            asyncUnaryCall(
              new MethodHandlers<
                io.grpc.examples.helloworld.HelloRequest,
                io.grpc.examples.helloworld.HelloReply>(
                  this, METHODID_SAY_HELLO)))
          .build();
    }

3.2.3 service 调用

gRPC 消息的接收

gRPC 消息的接入由 Netty HTTP/2 协议栈回调 gRPC 的 FrameListener,进而调用 NettyServerHandler 的 onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers) 和 onDataRead(int streamId, ByteBuf data, int padding, boolean endOfStream),代码如下所示:

消息头和消息体的处理,主要由 MessageDeframer 的 deliver 方法完成,相关代码如下(MessageDeframer 类):

if (inDelivery) {
     return;
   }
   inDelivery = true;
   try {
          while (pendingDeliveries > 0 && readRequiredBytes()) {
       switch (state) {
         case HEADER:
           processHeader();
           break;
         case BODY:
           processBody();
           pendingDeliveries--;
           break;
         default:
           throw new AssertionError("Invalid state: " + state);

gRPC 请求消息(PB)的解码由 PrototypeMarshaller 负责,代码如下 (ProtoLiteUtils 类):

public T parse(InputStream stream) {
       if (stream instanceof ProtoInputStream) {
         ProtoInputStream protoStream = (ProtoInputStream) stream;
         if (protoStream.parser() == parser) {
           try {
             T message = (T) ((ProtoInputStream) stream).message();
...
gRPC 响应消息发送

响应消息分为两部分发送:响应消息头和消息体,分别被封装成不同的 WriteQueue.AbstractQueuedCommand,插入到 WriteQueue 中。

消息头封装代码(NettyServerStream 类):

public void writeHeaders(Metadata headers) {
     writeQueue.enqueue(new SendResponseHeadersCommand(transportState(),
         Utils.convertServerHeaders(headers), false),
         true);
   }

消息体封装代码(NettyServerStream 类):

ByteBuf bytebuf = ((NettyWritableBuffer) frame).bytebuf();
     final int numBytes = bytebuf.readableBytes();
     onSendingBytes(numBytes);
     writeQueue.enqueue(
         new SendGrpcFrameCommand(transportState(), bytebuf, false),
         channel.newPromise().addListener(new ChannelFutureListener() {
           @Override
           public void operationComplete(ChannelFuture future) throws Exception {
             transportState().onSentBytes(numBytes);
           }
         }), flush);

Netty 的 NioEventLoop 将响应消息发送到 ChannelPipeline,最终被 NettyServerHandler 拦截并处理。

响应消息头处理代码如下(NettyServerHandler 类):

private void sendResponseHeaders(ChannelHandlerContext ctx, SendResponseHeadersCommand cmd,
     ChannelPromise promise) throws Http2Exception {
   int streamId = cmd.stream().id();
   Http2Stream stream = connection().stream(streamId);
   if (stream == null) {
     resetStream(ctx, streamId, Http2Error.CANCEL.code(), promise);
     return;
   }
   if (cmd.endOfStream()) {
     closeStreamWhenDone(promise, streamId);
   }
   encoder().writeHeaders(ctx, streamId, cmd.headers(), 0, cmd.endOfStream(), promise);
 }

响应消息体处理代码如下(NettyServerHandler 类):

private void sendGrpcFrame(ChannelHandlerContext ctx, SendGrpcFrameCommand cmd,
     ChannelPromise promise) throws Http2Exception {
   if (cmd.endStream()) {
     closeStreamWhenDone(promise, cmd.streamId());
   }
   encoder().writeData(ctx, cmd.streamId(), cmd.content(), 0, cmd.endStream(), promise);
 }

##### 服务接口实例调用: 经过一系列预处理,最终由 ServerCalls 的 ServerCallHandler 调用服务接口实例,代码如下(ServerCalls 类):

 return new EmptyServerCallListener<ReqT>() {
         ReqT request;
         @Override
         public void onMessage(ReqT request) {
           this.request = request;
         }
         @Override
         public void onHalfClose() {
           if (request != null) {
             method.invoke(request, responseObserver);
             responseObserver.freeze();
             if (call.isReady()) {
               onReady();
             }

最终的服务实现类调用如下(GreeterGrpc 类):

public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
     switch (methodId) {
       case METHODID_SAY_HELLO:
         serviceImpl.sayHello((io.grpc.examples.helloworld.HelloRequest) request,
 (io.grpc.stub.StreamObserver<io.grpc.examples.helloworld.HelloReply>) responseObserver);
         break;
       default:
         throw new AssertionError();
     }

3.3 服务端线程模型

gRPC 的线程由 Netty 线程 + gRPC 应用线程组成,它们之间的交互和切换比较复杂,下面做下详细介绍。

3.3.1 Netty Server 线程模型


它的工作流程总结如下:

Netty Server 使用的 NIO 线程实现是 NioEventLoop,它的职责如下:

3.3.2 gRPC service 线程模型

gRPC 服务端调度线程为 SerializingExecutor,它实现了 Executor 和 Runnable 接口,通过外部传入的 Executor 对象,调度和处理 Runnable,同时内部又维护了一个任务队列 ConcurrentLinkedQueue,通过 run 方法循环处理队列中存放的 Runnable 对象,代码示例如下:

3.3.3 线程调度和切换策略

Netty Server I/O 线程的职责:

gRPC service 线程的职责:

gRPC 的线程模型遵循 Netty 的线程分工原则,即:协议层消息的接收和编解码由 Netty 的 I/O(NioEventLoop) 线程负责;后续应用层的处理由应用线程负责,防止由于应用处理耗时而阻塞 Netty 的 I/O 线程。

基于上述分工原则,在 gRPC 请求消息的接入和响应发送过程中,系统不断的在 Netty I/O 线程和 gRPC 应用线程之间进行切换。明白了分工原则,也就能够理解为什么要做频繁的线程切换。

gRPC 线程模型存在的一个缺点,就是在一次 RPC 调用过程中,做了多次 I/O 线程到应用线程之间的切换,频繁切换会导致性能下降,这也是为什么 gRPC 性能比一些基于私有协议构建的 RPC 框架性能低的一个原因。尽管 gRPC 的性能已经比较优异,但是仍有一定的优化空间。

阅读原文

comments powered by Disqus
rss github weibo twitter instagram pinterest facebook linkedin stackoverflow reddit quora mail