spark2.1.0之源码分析——RPC服务器TransportServer

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

1. 《Spark2.1.0之内置RPC框架》
2. 《spark2.1.0之源码分析——RPC配置TransportConf》
3. 《spark2.1.0之源码分析——RPC客户端工厂TransportClientFactory》

       TransportServer是RPC框架的服务端，可提供高效的、低级别的流服务。在说明《Spark2.1.0之内置RPC框架》一文所展示的图1中的记号②时提到过TransportContext的createServer方法用于创建TransportServer，其实现见代码清单1。

代码清单1         创建RPC服务端

  public TransportServer createServer(int port, List<TransportServerBootstrap> bootstraps) {
    return new TransportServer(this, null, port, rpcHandler, bootstraps);
  }

  public TransportServer createServer(
      String host, int port, List<TransportServerBootstrap> bootstraps) {
    return new TransportServer(this, host, port, rpcHandler, bootstraps);
  }

  public TransportServer createServer(List<TransportServerBootstrap> bootstraps) {
    return createServer(0, bootstraps);
  }

  public TransportServer createServer() {
    return createServer(0, Lists.<TransportServerBootstrap>newArrayList());
  }

代码清单1中列出了四个名为createServer的重载方法，但是它们最终调用了TransportServer的构造器（见代码清单2）来创建TransportServer实例。

代码清单2         TransportServer的构造器

  public TransportServer(
      TransportContext context,
      String hostToBind,
      int portToBind,
      RpcHandler appRpcHandler,
      List<TransportServerBootstrap> bootstraps) {
    this.context = context;
    this.conf = context.getConf();
    this.appRpcHandler = appRpcHandler;
    this.bootstraps = Lists.newArrayList(Preconditions.checkNotNull(bootstraps));

    try {
      init(hostToBind, portToBind);
    } catch (RuntimeException e) {
      JavaUtils.closeQuietly(this);
      throw e;
    }
  }

TransportServer的构造器中的各个变量分别为：

• context：即参数传递的TransportContext的引用；
• conf：即TransportConf，这里通过调用TransportContext的getConf获取；
• appRpcHandler：即RPC请求处理器RpcHandler；
• bootstraps：即参数传递的TransportServerBootstrap列表；

TransportServer的构造器（见代码清单2）中调用了init方法，init方法用于对TransportServer进行初始化，见代码清单3。

代码清单3         TransportServer的初始化

  private void init(String hostToBind, int portToBind) {
    // 根据Netty的API文档，Netty服务端需同时创建bossGroup和workerGroup
    IOMode ioMode = IOMode.valueOf(conf.ioMode());
    EventLoopGroup bossGroup =
      NettyUtils.createEventLoop(ioMode, conf.serverThreads(), conf.getModuleName() + "-server");
    EventLoopGroup workerGroup = bossGroup;
    // 创建一个汇集ByteBuf但对本地线程缓存禁用的分配器
    PooledByteBufAllocator allocator = NettyUtils.createPooledByteBufAllocator(
      conf.preferDirectBufs(), true /* allowCache */, conf.serverThreads());
    // 创建Netty的服务端根引导程序并对其进行配置
    bootstrap = new ServerBootstrap()
      .group(bossGroup, workerGroup)
      .channel(NettyUtils.getServerChannelClass(ioMode))
      .option(ChannelOption.ALLOCATOR, allocator)
      .childOption(ChannelOption.ALLOCATOR, allocator);

    if (conf.backLog() > 0) {
      bootstrap.option(ChannelOption.SO_BACKLOG, conf.backLog());
    }
    if (conf.receiveBuf() > 0) {
      bootstrap.childOption(ChannelOption.SO_RCVBUF, conf.receiveBuf());
    }
    if (conf.sendBuf() > 0) {
      bootstrap.childOption(ChannelOption.SO_SNDBUF, conf.sendBuf());
    }
    // 为根引导程序设置管道初始化回调函数
    bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        RpcHandler rpcHandler = appRpcHandler;
        for (TransportServerBootstrap bootstrap : bootstraps) {
          rpcHandler = bootstrap.doBootstrap(ch, rpcHandler);
        }
        context.initializePipeline(ch, rpcHandler);
      }
    });
    // 给根引导程序绑定Socket的监听端口
    InetSocketAddress address = hostToBind == null ?
        new InetSocketAddress(portToBind): new InetSocketAddress(hostToBind, portToBind);
    channelFuture = bootstrap.bind(address);
    channelFuture.syncUninterruptibly();

    port = ((InetSocketAddress) channelFuture.channel().localAddress()).getPort();
    logger.debug("Shuffle server started on port: {}", port);
  }

代码清单3中TransportServer初始化的步骤如下：

1. 创建bossGroup和workerGroup（根据Netty的API文档，Netty服务端需同时创建bossGroup和workerGroup。）；

2. 创建一个汇集ByteBuf但对本地线程缓存禁用的分配器；

3. 调用Netty的API创建Netty的服务端根引导程序并对其进行配置；

4. 为根引导程序设置管道初始化回调函数，此回调函数首先设置TransportServerBootstrap到根引导程序中，然后调用TransportContext的initializePipeline方法初始化Channel的pipeline；

5. 给根引导程序绑定Socket的监听端口，最后返回监听的端口。

提示：代码清单3中使用了NettyUtils工具类的很多方法，在《附录G Netty与NettyUtils》中有对它们的详细介绍。EventLoopGroup、PooledByteBufAllocator、ServerBootstrap都是Netty提供的API，对于它们的更多介绍，请访问http://netty.io/。

关于《Spark内核设计的艺术 架构设计与实现》

经过近一年的准备，《Spark内核设计的艺术 架构设计与实现》一书现已出版发行，图书如图：

纸质版售卖链接如下：

京东：https://item.jd.com/12302500.html