CategoryResourceRepost/极客时间专栏/深入浅出gRPC/02 | 客户端创建和调用原理.md

1. gRPC客户端创建流程

1.1 背景

gRPC 是在 HTTP/2 之上实现的 RPC 框架，HTTP/2 是第7层（应用层）协议，它运行在TCP（第4层 - 传输层）协议之上，相比于传统的REST/JSON机制有诸多的优点：

1. 基于HTTP/2之上的二进制协议（Protobuf序列化机制）；
2. 一个连接上可以多路复用，并发处理多个请求和响应；
3. 多种语言的类库实现；
4. 服务定义文件和自动代码生成（.proto文件和Protobuf编译工具）。

此外，gRPC还提供了很多扩展点，用于对框架进行功能定制和扩展，例如，通过开放负载均衡接口可以无缝的与第三方组件进行集成对接（Zookeeper、域名解析服务、SLB服务等）。

一个完整的 RPC 调用流程示例如下：

gRPC的RPC调用与上述流程相似，下面我们一起学习下gRPC的客户端创建和服务调用流程。

1.2 业务代码示例

以gRPC入门级的helloworld Demo为例，客户端发起RPC调用的代码主要包括如下几部分：

1. 根据hostname和port创建ManagedChannelImpl；
2. 根据helloworld.proto文件生成的GreeterGrpc创建客户端Stub，用来发起RPC调用；
3. 使用客户端Stub（GreeterBlockingStub）发起RPC调用，获取响应。

相关示例代码如下所示（HelloWorldClient类）：

HelloWorldClient(ManagedChannelBuilder<?> channelBuilder) {
    channel = channelBuilder.build();
    blockingStub = GreeterGrpc.newBlockingStub(channel);
    futureStub = GreeterGrpc.newFutureStub(channel);
    stub = GreeterGrpc.newStub(channel);
  }
  public void blockingGreet(String name) {
    logger.info("Will try to greet " + name + " ...");
    HelloRequest request = HelloRequest.newBuilder().setName(name).build();
    try {
      HelloReply response = blockingStub
              .sayHello(request);
...

1.3 RPC调用流程

gRPC的客户端调用主要包括基于Netty的HTTP/2客户端创建、客户端负载均衡、请求消息的发送和响应接收处理四个流程。

1.3.1 客户端调用总体流程

gRPC的客户端调用总体流程如下图所示：

gRPC的客户端调用流程如下：

1. 客户端Stub(GreeterBlockingStub)调用sayHello(request)，发起RPC调用；
2. 通过DnsNameResolver进行域名解析，获取服务端的地址信息（列表），随后使用默认的LoadBalancer策略，选择一个具体的gRPC服务端实例；
3. 如果与路由选中的服务端之间没有可用的连接，则创建NettyClientTransport和NettyClientHandler，发起HTTP/2连接；
4. 对请求消息使用PB（Protobuf）做序列化，通过HTTP/2 Stream发送给gRPC服务端；
5. 接收到服务端响应之后，使用PB（Protobuf）做反序列化；
6. 回调GrpcFuture的set(Response)方法，唤醒阻塞的客户端调用线程，获取RPC响应。

需要指出的是，客户端同步阻塞RPC调用阻塞的是调用方线程（通常是业务线程），底层Transport的I/O线程（Netty的NioEventLoop）仍然是非阻塞的。

1.3.2 ManagedChannel创建流程

ManagedChannel是对Transport层SocketChannel的抽象，Transport层负责协议消息的序列化和反序列化，以及协议消息的发送和读取。

ManagedChannel将处理后的请求和响应传递给与之相关联的ClientCall进行上层处理，同时，ManagedChannel提供了对Channel的生命周期管理（链路创建、空闲、关闭等）。

ManagedChannel提供了接口式的切面ClientInterceptor，它可以拦截RPC客户端调用，注入扩展点，以及功能定制，方便框架的使用者对gRPC进行功能扩展。

ManagedChannel的主要实现类ManagedChannelImpl创建流程如下：

流程关键技术点解读：

1. 使用builder模式创建ManagedChannelBuilder实现类NettyChannelBuilder，NettyChannelBuilder提供了buildTransportFactory工厂方法创建NettyTransportFactory，最终用于创建NettyClientTransport；
2. 初始化HTTP/2连接方式：采用plaintext协商模式还是默认的TLS模式，HTTP/2的连接有两种模式，h2（基于TLS之上构建的HTTP/2）和h2c（直接在TCP之上构建的HTTP/2）；
3. 创建NameResolver.Factory工厂类，用于服务端URI的解析，gRPC默认采用DNS域名解析方式。

ManagedChannel实例构造完成之后，即可创建ClientCall，发起RPC调用。

1.3.3 ClientCall创建流程

完成ManagedChannelImpl创建之后，由ManagedChannelImpl发起创建一个新的ClientCall实例。ClientCall的用途是业务应用层的消息调度和处理，它的典型用法如下：

 call = channel.newCall(unaryMethod, callOptions);
 call.start(listener, headers);
 call.sendMessage(message);
 call.halfClose();
 call.request(1);
 // wait for listener.onMessage()

ClientCall实例的创建流程如下所示：

流程关键技术点解读：

ClientCallImpl的主要构造参数是MethodDescriptor和CallOptions，其中MethodDescriptor存放了需要调用RPC服务的接口名、方法名、服务调用的方式（例如UNARY类型）以及请求和响应的序列化和反序列化实现类。 CallOptions则存放了RPC调用的其它附加信息，例如超时时间、鉴权信息、消息长度限制和执行客户端调用的线程池等。

设置压缩和解压缩的注册类（CompressorRegistry和DecompressorRegistry），以便可以按照指定的压缩算法对HTTP/2消息做压缩和解压缩。

ClientCallImpl实例创建完成之后，就可以调用ClientTransport，创建HTTP/2 Client，向gRPC服务端发起远程服务调用。

1.3.4 基于Netty的HTTP/2 Client创建流程

gRPC客户端底层基于Netty4.1的HTTP/2协议栈框架构建，以便可以使用HTTP/2协议来承载RPC消息，在满足标准化规范的前提下，提升通信性能。

gRPC HTTP/2协议栈（客户端）的关键实现是NettyClientTransport和NettyClientHandler，客户端初始化流程如下所示：

流程关键技术点解读：

**NettyClientHandler的创建：**级联创建Netty的Http2FrameReader、Http2FrameWriter和Http2Connection，用于构建基于Netty的gRPC HTTP/2客户端协议栈。

**HTTP/2 Client启动：**仍然基于Netty的Bootstrap来初始化并启动客户端，但是有两个细节需要注意：

• NettyClientHandler（实际被包装成ProtocolNegotiator.Handler，用于HTTP/2的握手协商）创建之后，不是由传统的ChannelInitializer在初始化Channel时将NettyClientHandler加入到pipeline中，而是直接通过Bootstrap的handler方法直接加入到pipeline中，以便可以立即接收发送任务。
• 客户端使用的work线程组并非通常意义的EventLoopGroup，而是一个EventLoop：即HTTP/2客户端使用的work线程并非一组线程（默认线程数为CPU内核 * 2），而是一个EventLoop线程。这个其实也很容易理解，一个NioEventLoop线程可以同时处理多个HTTP/2客户端连接，它是多路复用的，对于单个HTTP/2客户端，如果默认独占一个work线程组，将造成极大的资源浪费，同时也可能会导致句柄溢出（并发启动大量HTTP/2客户端）。

**WriteQueue创建：**Netty的NioSocketChannel初始化并向Selector注册之后（发起HTTP连接之前），立即由NettyClientHandler创建WriteQueue，用于接收并处理gRPC内部的各种Command，例如链路关闭指令、发送Frame指令、发送Ping指令等。

HTTP/2 Client创建完成之后，即可由客户端根据协商策略发起HTTP/2连接。如果连接创建成功，后续即可复用该HTTP/2连接，进行RPC调用。

1.3.5 HTTP/2连接创建流程

HTTP/2在TCP连接之初通过协商的方式进行通信，只有协商成功，才能进行后续的业务层数据发送和接收。

HTTP/2的版本标识分为两类：

1. 基于TLS之上构架的HTTP/2, 即HTTPS，使用h2表示（ALPN）：0x68与0x32；
2. 直接在TCP之上构建的HTTP/2,即HTTP，使用h2c表示。

HTTP/2连接创建，分为两种：通过协商升级协议方式和直接连接方式。

假如不知道服务端是否支持HTTP/2，可以先使用HTTP/1.1进行协商，客户端发送协商请求消息（只含消息头），报文示例如下：

GET / HTTP/1.1
Host: 127.0.0.1
Connection: Upgrade, HTTP2-Settings
Upgrade: h2c
HTTP2-Settings: <base64url encoding of HTTP/2 SETTINGS payload>

服务端接收到协商请求之后，如果不支持HTTP/2，则直接按照HTTP/1.1响应返回，双方通过HTTP/1.1进行通信，报文示例如下：

HTTP/1.1 200 OK
Content-Length: 28
Content-Type: text/css

body...

如果服务端支持HTTP/2,则协商成功，返回101结果码，通知客户端一起升级到HTTP/2进行通信，示例报文如下：

HTTP/1.1 101 Switching Protocols
Connection: Upgrade
Upgrade: h2c

[ HTTP/2 connection...

101响应之后，服务需要发送SETTINGS帧作为连接序言，客户端接收到101响应之后，也必须发送一个序言作为回应，示例如下：

PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n
SETTINGS帧

客户端序言发送完成之后，可以不需要等待服务端的SETTINGS帧，而直接发送业务请求Frame。

假如客户端和服务端已经约定使用HTTP/2,则可以免去101协商和切换流程，直接发起HTTP/2连接，具体流程如下所示：

几个关键点：

1. 如果已经明确知道服务端支持HTTP/2，则可免去通过HTTP/1.1 101协议切换方式进行升级，TCP连接建立之后即可发送序言，否则只能在接收到服务端101响应之后发送序言；
2. 针对一个连接，服务端第一个要发送的帧必须是SETTINGS帧，连接序言所包含的SETTINGS帧可以为空；
3. 客户端可以在发送完序言之后发送应用帧数据，不用等待来自服务器端的序言SETTINGS帧。

gRPC支持三种Protocol Negotiator策略：

1. **PlaintextNegotiator：**明确服务端支持HTTP/2，采用HTTP直接连接的方式与服务端建立HTTP/2连接，省去101协议切换过程；
2. **PlaintextUpgradeNegotiator：**不清楚服务端是否支持HTTP/2，采用HTTP/1.1协商模式切换升级到HTTP/2；
3. **TlsNegotiator：**在TLS之上构建HTTP/2，协商采用ALPN扩展协议，以"h2"作为协议标识符。

下面我们以PlaintextNegotiator为例，了解下基于Netty的HTTP/2连接创建流程：

1.3.6 负载均衡策略

总体上看，RPC的负载均衡策略有两大类：

1. 服务端负载均衡（例如代理模式、外部负载均衡服务）
2. 客户端负载均衡（内置负载均衡策略和算法，客户端实现）

外部负载均衡模式如下所示：

以代理LB模式为例：RPC客户端向负载均衡代理发送请求，负载均衡代理按照指定的路由策略，将请求消息转发到后端可用的服务实例上。负载均衡代理负责维护后端可用的服务列表，如果发现某个服务不可用，则将其剔除出路由表。

代理LB模式的优点是客户端不需要实现负载均衡策略算法，也不需要维护后端的服务列表信息，不直接跟后端的服务进行通信，在做网络安全边界隔离时，非常实用。例如通过Nginx做L7层负载均衡，将互联网前端的流量安全的接入到后端服务中。

代理LB模式通常支持L4（Transport）和L7（Application)层负载均衡，两者各有优缺点，可以根据RPC的协议特点灵活选择。L4/L7层负载均衡对应场景如下：

• **L4层：**对时延要求苛刻、资源损耗少、RPC本身采用私有TCP协议；
• **L7层：**有会话状态的连接、HTTP协议簇（例如Restful）。

客户端负载均衡策略由客户端内置负载均衡能力，通过静态配置、域名解析服务（例如DNS服务）、订阅发布（例如Zookeeper服务注册中心）等方式获取RPC服务端地址列表，并将地址列表缓存到客户端内存中。

每次RPC调用时，根据客户端配置的负载均衡策略由负载均衡算法从缓存的服务地址列表中选择一个服务实例，发起RPC调用。

客户端负载均衡策略工作原理示例如下：

gRPC默认采用客户端负载均衡策略，同时提供了扩展机制，使用者通过自定义实现NameResolver和LoadBalancer，即可覆盖gRPC默认的负载均衡策略，实现自定义路由策略的扩展。

gRPC提供的负载均衡策略实现类如下所示：

• **PickFirstBalancer：**无负载均衡能力，即使有多个服务端地址可用，也只选择第一个地址；
• RoundRobinLoadBalancer："RoundRobin"负载均衡策略。

gRPC负载均衡流程如下所示：

流程关键技术点解读：

负载均衡功能模块的输入是客户端指定的hostName、需要调用的接口名和方法名等参数，输出是执行负载均衡算法后获得的NettyClientTransport，通过NettyClientTransport可以创建基于Netty HTTP/2的gRPC客户端，发起RPC调用；

gRPC系统默认提供的是DnsNameResolver，它通过InetAddress.getAllByName(host)获取指定host的IP地址列表（本地DNS服务），对于扩展者而言，可以继承NameResolver实现自定义的地址解析服务，例如使用Zookeeper替换DnsNameResolver，把Zookeeper作为动态的服务地址配置中心，它的伪代码示例如下： **第一步：**继承NameResolver，实现start(Listener listener)方法：

void start(Listener listener)
{
 //获取ZooKeeper地址，并连接
 //创建Watcher，并实现process(WatchedEvent event)，监听地址变更
 //根据接口名和方法名，调用getChildren方法，获取发布该服务的地址列表
//将地址列表加到List中
// 调用NameResolver.Listener.onAddresses(),通知地址解析完成

**第二步：**创建ManagedChannelBuilder时，指定Target的地址为Zookeeper服务端地址，同时设置nameResolver为Zookeeper NameResolver,示例代码如下所示：

this(ManagedChannelBuilder.forTarget(zookeeperAddr)
        .loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance())
        .nameResolverFactory(new ZookeeperNameResolverProvider())
        .usePlaintext(false));

1. LoadBalancer负责从nameResolver中解析获得的服务端URL中按照指定路由策略，选择一个目标服务端地址，并创建ClientTransport。同样，可以通过覆盖handleResolvedAddressGroups实现自定义负载均衡策略。

通过LoadBalancer + NameResolver，可以实现灵活的负载均衡策略扩展。例如基于Zookeeper、etcd的分布式配置服务中心方案。

1.3.7 RPC请求消息发送流程

gRPC默认基于Netty HTTP/2 + PB进行RPC调用，请求消息发送流程如下所示：

流程关键技术点解读：

1. ClientCallImpl的sendMessage调用，主要完成了请求对象的序列化（基于PB）、HTTP/2 Frame的初始化；
2. ClientCallImpl的halfClose调用将客户端准备就绪的请求Frame封装成自定义的SendGrpcFrameCommand，写入到WriteQueue中；
3. WriteQueue执行flush()将SendGrpcFrameCommand写入到Netty的Channel中，调用Channel的write方法，被NettyClientHandler拦截到，由NettyClientHandler负责具体的发送操作；
4. NettyClientHandler调用Http2ConnectionEncoder的writeData方法，将Frame写入到HTTP/2 Stream中，完成请求消息的发送。

1.3.8 RPC响应接收和处理流程

gRPC客户端响应消息的接收入口是NettyClientHandler，它的处理流程如下所示：

流程关键技术点解读：

1. NettyClientHandler的onHeadersRead(int streamId, Http2Headers headers, boolean endStream)方法会被调用两次，根据endStream判断是否是Stream结尾；
2. 请求和响应的关联：根据streamId可以关联同一个HTTP/2 Stream，将NettyClientStream缓存到Stream中，客户端就可以在接收到响应消息头或消息体时还原出NettyClientStream，进行后续处理；
3. RPC客户端调用线程的阻塞和唤醒使用到了GrpcFuture的wait和notify机制，来实现客户端调用线程的同步阻塞和唤醒；
4. 客户端和服务端的HTTP/2 Header和Data Frame解析共用同一个方法，即MessageDeframer的deliver()。

客户端源码分析

gRPC客户端调用原理并不复杂，但是代码却相对比较繁杂。下面围绕关键的类库，对主要功能点进行源码分析。

NettyClientTransport功能和源码分析

NettyClientTransport的主要功能如下：

• 通过start(Listener transportListener) 创建HTTP/2 Client，并连接gRPC服务端；
• 通过newStream(MethodDescriptor method, Metadata headers, CallOptions callOptions) 创建ClientStream；
• 通过shutdown() 关闭底层的HTTP/2连接。

以启动HTTP/2客户端为例进行讲解（NettyClientTransport类）：

EventLoop eventLoop = group.next();
    if (keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED) {
      keepAliveManager = new KeepAliveManager(
          new ClientKeepAlivePinger(this), eventLoop, keepAliveTimeNanos, keepAliveTimeoutNanos,
          keepAliveWithoutCalls);
    }
    handler = NettyClientHandler.newHandler(lifecycleManager, keepAliveManager, flowControlWindow,
        maxHeaderListSize, Ticker.systemTicker(), tooManyPingsRunnable);
    HandlerSettings.setAutoWindow(handler);
    negotiationHandler = negotiator.newHandler(handler);

根据启动时配置的HTTP/2协商策略，以NettyClientHandler为参数创建ProtocolNegotiator.Handler。

创建Bootstrap，并设置EventLoopGroup，需要指出的是，此处并没有使用EventLoopGroup，而是它的一种实现类EventLoop，原因在前文中已经说明，相关代码示例如下（NettyClientTransport类）：

Bootstrap b = new Bootstrap();
    b.group(eventLoop);
    b.channel(channelType);
    if (NioSocketChannel.class.isAssignableFrom(channelType)) {
      b.option(SO_KEEPALIVE, true);
    }

创建WriteQueue并设置到NettyClientHandler中，用于接收内部的各种QueuedCommand，初始化完成之后，发起HTTP/2连接，代码如下（NettyClientTransport类）：

handler.startWriteQueue(channel);
    channel.connect(address).addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (!future.isSuccess()) {
          ChannelHandlerContext ctx = future.channel().pipeline().context(handler);
          if (ctx != null) {
            ctx.fireExceptionCaught(future.cause());
          }
          future.channel().pipeline().fireExceptionCaught(future.cause());
        }

2.2 NettyClientHandler功能和源码分析

NettyClientHandler继承自Netty的Http2ConnectionHandler，是gRPC接收和发送HTTP/2消息的关键实现类，也是gRPC和Netty的交互桥梁，它的主要功能如下所示：

• 发送各种协议消息给gRPC服务端；
• 接收gRPC服务端返回的应答消息头、消息体和其它协议消息；
• 处理HTTP/2协议相关的指令，例如StreamError、ConnectionError等。

协议消息的发送：无论是业务请求消息，还是协议指令消息，都统一封装成QueuedCommand，由NettyClientHandler拦截并处理，相关代码如下所示（NettyClientHandler类）：

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
          throws Exception {
    if (msg instanceof CreateStreamCommand) {
      createStream((CreateStreamCommand) msg, promise);
    } else if (msg instanceof SendGrpcFrameCommand) {
      sendGrpcFrame(ctx, (SendGrpcFrameCommand) msg, promise);
    } else if (msg instanceof CancelClientStreamCommand) {
      cancelStream(ctx, (CancelClientStreamCommand) msg, promise);
    } else if (msg instanceof SendPingCommand) {
      sendPingFrame(ctx, (SendPingCommand) msg, promise);
    } else if (msg instanceof GracefulCloseCommand) {
      gracefulClose(ctx, (GracefulCloseCommand) msg, promise);
    } else if (msg instanceof ForcefulCloseCommand) {
      forcefulClose(ctx, (ForcefulCloseCommand) msg, promise);
    } else if (msg == NOOP_MESSAGE) {
      ctx.write(Unpooled.EMPTY_BUFFER, promise);
    } else {
      throw new AssertionError("Write called for unexpected type: " + msg.getClass().getName());
    }

协议消息的接收：NettyClientHandler通过向Http2ConnectionDecoder注册FrameListener来监听RPC响应消息和协议指令消息，相关接口如下：

FrameListener回调NettyClientHandler的相关方法，实现协议消息的接收和处理：

需要指出的是，NettyClientHandler 并没有实现所有的回调接口，对于需要特殊处理的几个方法进行了重载，例如onDataRead和onHeadersRead。

2.3 ProtocolNegotiator功能和源码分析

ProtocolNegotiator用于HTTP/2连接创建的协商，gRPC支持三种策略并有三个实现子类：

gRPC的ProtocolNegotiator实现类完全遵循HTTP/2相关规范，以PlaintextUpgradeNegotiator为例，通过设置Http2ClientUpgradeCodec，用于101协商和协议升级，相关代码如下所示（PlaintextUpgradeNegotiator类）：

public Handler newHandler(GrpcHttp2ConnectionHandler handler) {
      Http2ClientUpgradeCodec upgradeCodec = new Http2ClientUpgradeCodec(handler);
      HttpClientCodec httpClientCodec = new HttpClientCodec();
      final HttpClientUpgradeHandler upgrader =
          new HttpClientUpgradeHandler(httpClientCodec, upgradeCodec, 1000);
      return new BufferingHttp2UpgradeHandler(upgrader);
    }

2.4 LoadBalancer功能和源码分析

LoadBalancer负责客户端负载均衡，它是个抽象类，gRPC框架的使用者可以通过继承的方式进行扩展。

gRPC当前已经支持PickFirstBalancer和RoundRobinLoadBalancer两种负载均衡策略，未来不排除会提供更多的策略。

以RoundRobinLoadBalancer为例，它的工作原理如下：根据PickSubchannelArgs来选择一个Subchannel（RoundRobinLoadBalancerFactory类）：

public PickResult pickSubchannel(PickSubchannelArgs args) {
      if (size > 0) {
        return PickResult.withSubchannel(nextSubchannel());
      }
      if (status != null) {
        return PickResult.withError(status);
      }
      return PickResult.withNoResult();
    }

再看下Subchannel的选择算法（Picker类）：

private Subchannel nextSubchannel() {
      if (size == 0) {
        throw new NoSuchElementException();
      }
      synchronized (this) {
        Subchannel val = list.get(index);
        index++;
        if (index >= size) {
          index = 0;
        }
        return val;
      }
    }

即通过顺序的方式从服务端列表中获取一个Subchannel。
如果用户需要定制负载均衡策略，则可以在RPC调用时，使用如下代码（HelloWorldClient类）：

this(ManagedChannelBuilder.forAddress(host, port).loadBalancerFactory(RoundRobinLoadBalancerFactory.getInstance()).nameResolverFactory(new ZkNameResolverProvider()) .usePlaintext(true));

2.5 ClientCalls功能和源码分析

ClientCalls提供了各种RPC调用方式，包括同步、异步、Streaming和Unary方式等，相关方法如下所示：

下面一起看下RPC请求消息的发送和应答接收相关代码。

2.5.1 RPC请求调用源码分析

请求调用主要有两步：请求Frame构造和Frame发送，请求Frame构造代码如下所示（ClientCallImpl类）：

public void sendMessage(ReqT message) {
    Preconditions.checkState(stream != null, "Not started");
    Preconditions.checkState(!cancelCalled, "call was cancelled");
    Preconditions.checkState(!halfCloseCalled, "call was half-closed");
    try {
      InputStream messageIs = method.streamRequest(message);
      stream.writeMessage(messageIs);
...

使用PB对请求消息做序列化，生成InputStream，构造请求Frame:

private int writeUncompressed(InputStream message, int messageLength) throws IOException {
    if (messageLength != -1) {
      statsTraceCtx.outboundWireSize(messageLength);
      return writeKnownLengthUncompressed(message, messageLength);
    }
    BufferChainOutputStream bufferChain = new BufferChainOutputStream();
    int written = writeToOutputStream(message, bufferChain);
    if (maxOutboundMessageSize >= 0 && written > maxOutboundMessageSize) {
      throw Status.INTERNAL
          .withDescription(
              String.format("message too large %d > %d", written , maxOutboundMessageSize))
          .asRuntimeException();
    }
    writeBufferChain(bufferChain, false);
    return written;
}

Frame发送代码如下所示：

public void writeFrame(WritableBuffer frame, boolean endOfStream, boolean flush) {
      ByteBuf bytebuf = frame == null ? EMPTY_BUFFER : ((NettyWritableBuffer) frame).bytebuf();
      final int numBytes = bytebuf.readableBytes();
      if (numBytes > 0) {
        onSendingBytes(numBytes);
        writeQueue.enqueue(
            new SendGrpcFrameCommand(transportState(), bytebuf, endOfStream),
            channel.newPromise().addListener(new ChannelFutureListener() {
              @Override
              public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                  transportState().onSentBytes(numBytes);
                }
              }
            }), flush);

NettyClientHandler接收到发送事件之后，调用Http2ConnectionEncoder将Frame写入Netty HTTP/2协议栈（NettyClientHandler类）：

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

2.5.2 RPC响应接收和处理源码分析

响应消息的接收入口是NettyClientHandler，包括HTTP/2 Header和HTTP/2 DATA Frame两部分，代码如下（NettyClientHandler类）：

private void onHeadersRead(int streamId, Http2Headers headers, boolean endStream) {
    NettyClientStream.TransportState stream = clientStream(requireHttp2Stream(streamId));
    stream.transportHeadersReceived(headers, endStream);
    if (keepAliveManager != null) {
      keepAliveManager.onDataReceived();
    }
  }

如果参数endStream为True，说明Stream已经结束，调用transportTrailersReceived，通知Listener close，代码如下所示（AbstractClientStream2类）：

if (stopDelivery || isDeframerStalled()) {
        deliveryStalledTask = null;
        closeListener(status, trailers);
      } else {
        deliveryStalledTask = new Runnable() {
          @Override
          public void run() {
            closeListener(status, trailers);
          }
        };
      }

读取到HTTP/2 DATA Frame之后，调用MessageDeframer的deliver对Frame进行解析，代码如下（MessageDeframer类）：

private void deliver() {
    if (inDelivery) {
      return;
    }
    inDelivery = true;
    try {
          while (pendingDeliveries > 0 && readRequiredBytes()) {
        switch (state) {
          case HEADER:
            processHeader();
            break;
          case BODY:
            processBody();
...

将Frame 转换成InputStream之后，通知ClientStreamListenerImpl，调用messageRead(final InputStream message)，将InputStream反序列化为响应对象，相关代码如下所示（ClientStreamListenerImpl类）：

public void messageRead(final InputStream message) {
      class MessageRead extends ContextRunnable {
        MessageRead() {
          super(context);
        }
        @Override
        public final void runInContext() {
          try {
            if (closed) {
              return;
            }
            try {
              observer.onMessage(method.parseResponse(message));
            } finally {
              message.close();
            }

当接收到endOfStream之后，通知ClientStreamListenerImpl，调用它的close方法，如下所示（ClientStreamListenerImpl类）：

private void close(Status status, Metadata trailers) {
      closed = true;
      cancelListenersShouldBeRemoved = true;
      try {
        closeObserver(observer, status, trailers);
      } finally {
        removeContextListenerAndCancelDeadlineFuture();
      }
    }

最终调用UnaryStreamToFuture的onClose方法，set响应对象，唤醒阻塞的调用方线程，完成RPC调用，代码如下（UnaryStreamToFuture类）：

public void onClose(Status status, Metadata trailers) {
      if (status.isOk()) {
        if (value == null) {
          responseFuture.setException(
              Status.INTERNAL.withDescription("No value received for unary call")
                  .asRuntimeException(trailers));
        }
        responseFuture.set(value);
      } else {
        responseFuture.setException(status.asRuntimeException(trailers));
      }

源代码下载地址：

链接: https://github.com/geektime-geekbang/gRPC_LLF/tree/master