Alluxio源码分析:RPC框架浅析(二)-阿里云开发者社区

开发者社区> 数据库> 正文

Alluxio源码分析:RPC框架浅析(二)

简介:         Alluxio源码分析是一个基于内存的分布式文件系统,和HDFS、HBase等一样,也是由主从节点构成的。而节点之间的通信,一般都是采用的RPC通讯模型。Alluxio中RPC是基于何种技术如何实现的呢?它对于RPC请求是如何处理的?都涉及到哪些组件?本文将针对这些问题,为您一一解答。

        Alluxio源码分析是一个基于内存的分布式文件系统,和HDFS、HBase等一样,也是由主从节点构成的。而节点之间的通信,一般都是采用的RPC通讯模型。Alluxio中RPC是基于何种技术如何实现的呢?它对于RPC请求是如何处理的?都涉及到哪些组件?本文将针对这些问题,为您一一解答。

        继《Alluxio源码分析:RPC框架浅析(一)》一文后,本文继续讲解Alluxio中RPC实现。

        3、Server端实现:RPC Server端口绑定、传输协议等参数设置、Server启动

        AlluxioMaster是Alluxio中Master的实现,那么RPC服务端server自然就会落在它身上了。我们先看AlluxioMaster进程的启动main()方法,如下:

  /**
   * Starts the Alluxio master server via {@code java -cp <ALLUXIO-VERSION> alluxio.Master}.
   *
   * @param args there are no arguments used
   */
  public static void main(String[] args) {
    
	// 启动master时参数应为空
	if (args.length != 0) {
      LOG.info("java -cp {} alluxio.Master", Version.ALLUXIO_JAR);
      System.exit(-1);
    }

    // validate the conf
	// 验证配置信息
    if (!ValidateConf.validate()) {
      LOG.error("Invalid configuration found");
      System.exit(-1);
    }

    try {
    	
      // 调用get()方法,返回AlluxioMaster实例master
      AlluxioMaster master = get();
      
      // 调用实例master的start()方法,启动AlluxioMaster实例master
      master.start();
    } catch (Exception e) {
      LOG.error("Uncaught exception terminating Master", e);
      System.exit(-1);
    }
  }
        它主要干了两件事,一个就是调用get()方法,返回AlluxioMaster实例master,另一个就是调用实例master的start()方法,启动AlluxioMaster实例master。我们先看下get()方法,如下:

  /**
   * Returns a handle to the Alluxio master instance.
   *
   * @return Alluxio master handle
   */
  public static synchronized AlluxioMaster get() {
    
	// 静态AlluxioMaster类型成员变量sAlluxioMaster为空时,通过Factory.create()构造一个,否则返回sAlluxioMaster
	if (sAlluxioMaster == null) {
      sAlluxioMaster = Factory.create();
    }
    return sAlluxioMaster;
  }
        而Factory的create()方法,则会根据参数alluxio.zookeeper.enabled确定返回FaultTolerantAlluxioMaster实例还是AlluxioMaster实例,FaultTolerantAlluxioMaster继承自AlluxioMaster,默认是返回AlluxioMaster实例,代码如下:

    /**
     * @return {@link FaultTolerantAlluxioMaster} if Alluxio configuration is set to use zookeeper,
     *         otherwise, return {@link AlluxioMaster}.
     */
    public static AlluxioMaster create() {
    	
      // 根据参数alluxio.zookeeper.enabled确定返回FaultTolerantAlluxioMaster实例还是AlluxioMaster实例,
      // FaultTolerantAlluxioMaster继承自AlluxioMaster,默认是返回AlluxioMaster实例
      if (MasterContext.getConf().getBoolean(Constants.ZOOKEEPER_ENABLED)) {
        return new FaultTolerantAlluxioMaster();
      }
      return new AlluxioMaster();
    }
        在AlluxioMasterd的构造方法中,涉及RPC相关的,主要是Worker最大和最小线程数、服务端Socker的TServerSocket实例mTServerSocket等的构造,关键代码如下:

    Configuration conf = MasterContext.getConf();

    // Worker最大和最小线程数:分别取参数alluxio.master.worker.threads.max和alluxio.master.worker.threads.min
    mMinWorkerThreads = conf.getInt(Constants.MASTER_WORKER_THREADS_MIN);
    mMaxWorkerThreads = conf.getInt(Constants.MASTER_WORKER_THREADS_MAX);

      // 获取传输提供者mTransportProvider
      mTransportProvider = TransportProvider.Factory.create(conf);
      
      // 构造TServerSocket实例mTServerSocket
      // port取参数alluxio.master.port=19998
      mTServerSocket =
          new TServerSocket(NetworkAddressUtils.getBindAddress(ServiceType.MASTER_RPC, conf));

        再看实例master的start()方法,也就是AlluxioMaster的start()方法,代码如下:

  /**
   * Starts the Alluxio master server.
   *
   * @throws Exception if starting the master fails
   */
  public void start() throws Exception {
    startMasters(true);
    
    // 启动服务
    startServing();
  }
        继续看startServing()方法,如下:

  protected void startServing(String startMessage, String stopMessage) {
    mMasterMetricsSystem.start();
    
    // 启动web服务
    startServingWebServer();
    LOG.info("Alluxio Master version {} started @ {} {}", Version.VERSION, mMasterAddress,
        startMessage);
    
    // 启动RPC服务
    startServingRPCServer();
    LOG.info("Alluxio Master version {} ended @ {} {}", Version.VERSION, mMasterAddress,
        stopMessage);
  }
        撇开启动web服务不说,我们看下启动RPC服务的startServingRPCServer()方法,如下:

  protected void startServingRPCServer() {
    // set up multiplexed thrift processors
	// 构造多路复用处理器TMultiplexedProcessor实例processor
    TMultiplexedProcessor processor = new TMultiplexedProcessor();
    
    // 注册BlockMaster服务
    registerServices(processor, mBlockMaster.getServices());
    
    // 注册FileSystemMaster服务
    registerServices(processor, mFileSystemMaster.getServices());
    
    // 必要的话,注册LineageMaster服务
    if (LineageUtils.isLineageEnabled(MasterContext.getConf())) {
      registerServices(processor, mLineageMaster.getServices());
    }
    
    // register additional masters for RPC service
    // 注册额外的Masters服务
    for (Master master : mAdditionalMasters) {
      registerServices(processor, master.getServices());
    }

    // Return a TTransportFactory based on the authentication type
    TTransportFactory transportFactory;
    try {
    	
      // 获得传输工厂实例
      transportFactory = mTransportProvider.getServerTransportFactory();
    } catch (IOException e) {
      throw Throwables.propagate(e);
    }

    // create master thrift service with the multiplexed processor.
    
    // 构造TThreadPoolServer实例时需要的参数:
    // 服务端Socker:TServerSocket类型实例mTServerSocket
    // 最大Worker线程数mMaxWorkerThreads
    // 最小Worker线程数mMinWorkerThreads
    // 处理器processor
    // 传输工厂transportFactory
    // 协议工厂TBinaryProtocol:二进制协议TBinaryProtocol
    Args args = new TThreadPoolServer.Args(mTServerSocket).maxWorkerThreads(mMaxWorkerThreads)
        .minWorkerThreads(mMinWorkerThreads).processor(processor).transportFactory(transportFactory)
        .protocolFactory(new TBinaryProtocol.Factory(true, true));
    if (MasterContext.getConf().getBoolean(Constants.IN_TEST_MODE)) {
      args.stopTimeoutVal = 0;
    } else {
      args.stopTimeoutVal = Constants.THRIFT_STOP_TIMEOUT_SECONDS;
    }
    
    // 构造TThreadPoolServer实例mMasterServiceServer
    mMasterServiceServer = new TThreadPoolServer(args);

    // start thrift rpc server
    // 标志位正在提供服务mIsServing设置为true
    mIsServing = true;
    // 启动时间mStartTimeMs取当前时间
    mStartTimeMs = System.currentTimeMillis();
    
    // 启动TThreadPoolServer服务
    mMasterServiceServer.serve();
  }
         startServingRPCServer()方法主要处理流程如下:

        1、构造多路复用处理器TMultiplexedProcessor实例processor;

        2、调用registerServices()方法,注册BlockMaster、FileSystemMaster、LineageMaster、额外的Masters等各种服务;

        3、调用TransportProvider的getServerTransportFactory()方法获得传输工厂实例transportFactory;

        4、构造TThreadPoolServer实例时需要的参数:包括:

             (1)服务端Socker:TServerSocket类型实例mTServerSocket;

             (2)最大Worker线程数mMaxWorkerThreads;

             (3)最小Worker线程数mMinWorkerThreads;

             (4)多路复用处理器processor;

             (5)传输工厂transportFactory;

             (6)协议工厂TBinaryProtocol:二进制协议TBinaryProtocol;

        5、构造TThreadPoolServer实例mMasterServiceServer;

        6、标志位正在提供服务mIsServing设置为true;

        7、启动时间mStartTimeMs取当前时间;

        8、调用mMasterServiceServer的serve()方法启动TThreadPoolServer服务。
        先以FileSystemMaster服务为例,看下RPC服务是如何注册的,代码如下:

  private void registerServices(TMultiplexedProcessor processor, Map<String, TProcessor> services) {
    for (Map.Entry<String, TProcessor> service : services.entrySet()) {
      processor.registerProcessor(service.getKey(), service.getValue());
    }
  }

        注册很简单,多路复用处理器processor的registerProcessor()方法即可完成注册,关键是要看注册的是什么东西它的服务是通过FileSystemMaster的getServices()方法获取的,我们跟踪下:

  @Override
  public Map<String, TProcessor> getServices() {
    Map<String, TProcessor> services = new HashMap<String, TProcessor>();
    
    // FileSystemMasterClientService服务
    services.put(
    		
        // key为"FileSystemMasterClient"
        Constants.FILE_SYSTEM_MASTER_CLIENT_SERVICE_NAME,
        
        // 可以看出,FileSystemMasterClientService服务Master端实现者是FileSystemMasterClientServiceHandler
        new FileSystemMasterClientService.Processor<FileSystemMasterClientServiceHandler>(
            new FileSystemMasterClientServiceHandler(this)));
    
    // FileSystemMasterWorkerService服务  
    services.put(
    		
        // key为"FileSystemMasterWorker"
        Constants.FILE_SYSTEM_MASTER_WORKER_SERVICE_NAME,
        
        // 可以看出,FileSystemMasterWorkerService服务Master端实现者是FileSystemMasterWorkerServiceHandler
        new FileSystemMasterWorkerService.Processor<FileSystemMasterWorkerServiceHandler>(
            new FileSystemMasterWorkerServiceHandler(this)));
    return services;
  }
        就俩服务:FileSystemMasterClientService和FileSystemMasterWorkerService,它们在Master端的实现者分别是FileSystemMasterClientServiceHandler和FileSystemMasterWorkerServiceHandler,并且是通过各自Service的Processor来构造的,看到这里,你似乎应该明白什么了吧!这就是Processor的用途。
        再看下获得传输工厂实例transportFactory,它是通过TransportProvider实例mTransportProvider的getServerTransportFactory()方法来获取的,而mTransportProvider的初始化也是在AlluxioMaster构造方法中,通过TransportProvider.Factory.create(conf)来获取的,我们看下代码:

    public static TransportProvider create(Configuration conf) {
      AuthType authType = conf.getEnum(Constants.SECURITY_AUTHENTICATION_TYPE, AuthType.class);
      switch (authType) {
        case NOSASL:
          return new NoSaslTransportProvider(conf);
        case SIMPLE: // intended to fall through
        case CUSTOM:
          return new PlainSaslTransportProvider(conf);
        case KERBEROS:
          throw new UnsupportedOperationException(
              "getClientTransport: Kerberos is not supported currently.");
        default:
          throw new UnsupportedOperationException(
              "getClientTransport: Unsupported authentication type: " + authType.getAuthName());
      }
    }
        它目前仅支持NOSASL和CUSTOM两种类型,分别对应NoSaslTransportProvider和PlainSaslTransportProvider两个类。我们以CUSTOM类型的PlainSaslTransportProvider为例来看下getServerTransportFactory()方法,代码如下:

  @Override
  public TTransportFactory getServerTransportFactory() throws SaslException {
    AuthType authType =
        mConfiguration.getEnum(Constants.SECURITY_AUTHENTICATION_TYPE, AuthType.class);
    TSaslServerTransport.Factory saslFactory = new TSaslServerTransport.Factory();
    AuthenticationProvider provider =
        AuthenticationProvider.Factory.create(authType, mConfiguration);
    saslFactory
        .addServerDefinition(PlainSaslServerProvider.MECHANISM, null, null,
            new HashMap<String, String>(), new PlainSaslServerCallbackHandler(provider));
    return saslFactory;
  }
        剩余的TThreadPoolServer实例构造、参数选择等上面解释的已经很清晰,读者可自行分析。

        未完待续,请关注《Alluxio源码分析:RPC框架浅析(三)》









版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
数据库
使用钉钉扫一扫加入圈子
+ 订阅

分享数据库前沿,解构实战干货,推动数据库技术变革

其他文章