spark2.1.0之源码分析——RPC服务端引导程序TransportServerBootstrap

提示：阅读本文前最好先阅读：

《Spark2.1.0之内置RPC框架》
《spark2.1.0之源码分析——RPC配置TransportConf》
《spark2.1.0之源码分析——RPC客户端工厂TransportClientFactory》
《spark2.1.0之源码分析——RPC服务器TransportServer》
《spark2.1.0之源码分析——RPC管道初始化》
《spark2.1.0之源码分析——RPC传输管道处理器详解》
《spark2.1.0之源码分析——服务端RPC处理器RpcHandler详解》

通过《spark2.1.0之源码分析——RPC服务器TransportServer》一文的介绍，我们知道TransportServer的构造器中的bootstraps是TransportServerBootstrap的列表。接口TransportServerBootstrap定义了服务端引导程序的规范，服务端引导程序旨在当客户端与服务端建立连接之后，在服务端持有的客户端管道上执行的引导程序。TransportServerBootstrap的定义见代码清单1。

代码清单1 TransportServerBootstrap的定义

public interface TransportServerBootstrap {
  RpcHandler doBootstrap(Channel channel, RpcHandler rpcHandler);
}

TransportServerBootstrap的doBootstrap方法将对服务端的RpcHandler进行代理，接收客户端的请求。TransportServerBootstrap有SaslServerBootstrap和EncryptionCheckerBootstrap两个实现类。为了更清楚的说明TransportServerBootstrap的意义，我们以SaslServerBootstrap为例，来讲解其实现（见代码清单2）。

代码清单2 SaslServerBootstrap的doBootstrap实现

  public RpcHandler doBootstrap(Channel channel, RpcHandler rpcHandler) {
    return new SaslRpcHandler(conf, channel, rpcHandler, secretKeyHolder);
  }

根据代码清单2，我们知道SaslServerBootstrap的doBootstrap方法实际创建了SaslRpcHandler，SaslRpcHandler负责对管道进行SASL（Simple Authentication and Security Layer）加密。SaslRpcHandler本身也继承了RpcHandler，所以我们重点来看其receive方法的实现，见代码清单3。

代码清单3 SaslRpcHandler的receive方法

  @Override
  public void receive(TransportClient client, ByteBuffer message, RpcResponseCallback callback) {
    if (isComplete) {
      // 将消息传递给SaslRpcHandler所代理的下游RpcHandler并返回
      delegate.receive(client, message, callback);
      return;
    }

    ByteBuf nettyBuf = Unpooled.wrappedBuffer(message);
    SaslMessage saslMessage;
    try {
      saslMessage = SaslMessage.decode(nettyBuf);// 对客户端发送的消息进行SASL解密
    } finally {
      nettyBuf.release();
    }

    if (saslServer == null) {
      // 如果saslServer还未创建，则需要创建SparkSaslServer
      client.setClientId(saslMessage.appId);
      saslServer = new SparkSaslServer(saslMessage.appId, secretKeyHolder,
        conf.saslServerAlwaysEncrypt());
    }

    byte[] response;
    try {
      response = saslServer.response(JavaUtils.bufferToArray(// 使用saslServer处理已解密的消息
        saslMessage.body().nioByteBuffer()));
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }
    callback.onSuccess(ByteBuffer.wrap(response));

    if (saslServer.isComplete()) {
      logger.debug("SASL authentication successful for channel {}", client);
      isComplete = true;// SASL认证交换已经完成
      if (SparkSaslServer.QOP_AUTH_CONF.equals(saslServer.getNegotiatedProperty(Sasl.QOP))) {
        logger.debug("Enabling encryption for channel {}", client);
        // 对管道进行SASL加密
        SaslEncryption.addToChannel(channel, saslServer, conf.maxSaslEncryptedBlockSize());
        saslServer = null;
      } else {
        saslServer.dispose();
        saslServer = null;
      }
    }
  }

根据代码清单3，SaslRpcHandler处理客户端消息的步骤如下：

如果SASL认证交换已经完成（isComplete等于true），则将消息传递给SaslRpcHandler所代理的下游RpcHandler并返回。
如果SASL认证交换未完成（isComplete等于false），则对客户端发送的消息进行SASL解密。
如果saslServer还未创建，则需要创建SparkSaslServer。当SaslRpcHandler接收到客户端的第一条消息时会做此操作。
使用saslServer处理已解密的消息，并将处理结果通过RpcResponseCallback的回调方法返回给客户端。
如果SASL认证交换已经完成，则将isComplete置为true。
对管道进行SASL加密。

SaslServerBootstrap是通过SaslRpcHandler对下游RpcHandler进行代理的一种TransportServerBootstrap。EncryptionCheckerBootstrap是另一种TransportServerBootstrap的实现，它通过将自身加入Netty的管道中实现引导，EncryptionCheckerBootstrap的doBootstrap方法的实现见代码清单4。

代码清单4 EncryptionCheckerBootstrap的doBootstrap实现

    @Override
    public RpcHandler doBootstrap(Channel channel, RpcHandler rpcHandler) {
      channel.pipeline().addFirst("encryptionChecker", this);
      return rpcHandler;
    }

在详细介绍了TransportChannelHandler之后我们就可以对《spark2.1.0之源码分析——RPC管道初始化》文中的图1进行扩展，把TransportRequestHandler、TransportServerBootstrap及RpcHandler的处理流程增加进来，如下图所示。

RPC框架服务端处理请求、响应流程图