image.png

消息头和消息体的处理，主要由 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();
     }

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

Netty Server 线程模型

image.png

它的工作流程总结如下：

• 从主线程池（bossGroup）中随机选择一个 Reactor 线程作为 Acceptor 线程，用于绑定监听端口，接收客户端连接；
• Acceptor 线程接收客户端连接请求之后创建新的 SocketChannel，将其注册到主线程池（bossGroup）的其它 Reactor 线程上，由其负责接入认证、握手等操作；
• 步骤 2 完成之后，应用层的链路正式建立，将 SocketChannel 从主线程池的 Reactor 线程的多路复用器上摘除，重新注册到 Sub 线程池（workerGroup）的线程上，用于处理 I/O 的读写操作。

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

作为服务端 Acceptor 线程，负责处理客户端的请求接入；
作为客户端 Connecor 线程，负责注册监听连接操作位，用于判断异步连接结果；
作为 I/O 线程，监听网络读操作位，负责从 SocketChannel 中读取报文；
作为 I/O 线程，负责向 SocketChannel 写入报文发送给对方，如果发生写半包，会自动注册监听写事件，用于后续继续发送半包数据，直到数据全部发送完成；
作为定时任务线程，可以执行定时任务，例如链路空闲检测和发送心跳消息等；
作为线程执行器可以执行普通的任务 Task（Runnable）。

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

线程调度和切换策略
Netty Server I/O 线程的职责：

• gRPC 请求消息的读取、响应消息的发送

• HTTP/2 协议消息的编码和解码

• NettyServerHandler 的调度
  gRPC service 线程的职责：

• 将 gRPC 请求消息（PB 码流）反序列化为接口的请求参数对象

• 将接口响应对象序列化为 PB 码流

• gRPC 服务端接口实现类调用

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

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

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