Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 [三] 概述&原理

我不是女神ヾ 2023-05-21 12:51 84阅读 0赞

# 1.概述 #

在这里RPC实现其实就是分三部分, 分别是 协议定义&实现 , Server端实现和Client实现. 三个部分. 下面会分别进行讲述

# 2.协议实现 #

## 2.1.定义协议 ##

其实就是根据业务需要定义一个借口协议.

协议详情可以参看:  [Hadoop3.2.1 【 HDFS 】源码分析 : RPC实现 \[一\] proto接口协议][Hadoop3.2.1 _ HDFS _ _ RPC_ _ proto]

示例如下:

/**
     * 协议接口
     */
    public interface ClicentNameNodeProtocol {
      //1. 定义协议的ID
      public static final long versionID = 1L;
      /**
       * 拿到元数据方法，协议通信前会访问元数据
       */
      public String getMetaData(String path);
    }

## 2.2.实现协议 ##

实现协议这个也很简单, 就是根据接口创建一个实现类. 用的时候注册到Server服务中即可.

示例如下:

/**
     * 实现协议结构
     */
    public class ClicentNameNodeImpl implements ClicentNameNodeProtocol {
      public String getMetaData(String path) {
        // 数据存放的路径，有多少块，块大小，校验和，存储在哪一台机器上
        return path + ":3 - {BLOCK_1,BLOCK_2,BLOCK_3....";
      }
    }

整理的调用逻辑图,参考如下. 后面来具体分析.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70][]

## 3. RPC Server处理流程 ##

## 3.1.Server架构图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1][]

## 3.2.Server组件讲解 ##

> ■ **Listener**：   
>     Listener对象中存在一个Selector对象acceptSelector， 负责监听来自客户端的Socket连接请求。 当acceptSelector监听到连接请求后， Listener对象会初始化这个连接， 之后采用轮询的方式从readers线程池中选出一个Reader线程处理RPC请求的读取操作。
> 
> **■ Reader**：   
>      用于读取RPC请求。 Reader线程类中存在一个Selector对象readSelector， 类似于Reactor模式中的readReactor， 这个对象用于监听网络中是否有可以读取的RPC请求。 当readSelector监听到有可读的RPC请求后， 会唤醒Reader线程读取这个请求， 并将请求封装在一个Call对象中， 然后将这个Call对象放入共享队列CallQueue中。
> 
> **■ Handler**：   
>     用于处理RPC请求并发回响应。Handler对象会从CallQueue中不停地取出RPC请求， 然后执行RPC请求对应的本地函数， 最后封装响应并将响应发回客户端。 为了能够并发地处理RPC请求，Server中会存在多个Handler对象。
> 
> **■ Responder：**   
>     用于向客户端发送RPC响应，  响应很大或者网络条件不佳等情况下， Handler线程很难将完整的响应发回客户端， 这就会造成Handler线程阻塞， 从而影响RPC请求的处理效率。 所以Handler在没能够将完整的RPC响应发回客户端时， 会在Responder内部的respondSelector上注册一个写响应事件， 这里的respondSelector与Reactor模式的respondSelector概念相同， 当respondSelector监听到网络情况具备写响应的条件时， 会通知Responder将剩余响应发回客户端。

# 3.2.Server组件代码讲解 #

Server服务是用RPC.Builder类中的build()方法进行构建的. 下面是构建Server的模拟代码.

构建server的模拟代码:

public class Server {
      public static void main(String[] args) throws IOException {
        //1. 构建RPC框架
        RPC.Builder builder = new RPC.Builder(new Configuration());
        //2. 绑定地址
        builder.setBindAddress("localhost");
        //3. 绑定端口
        builder.setPort(7777);
        //4. 绑定协议
        builder.setProtocol(ClicentNameNodeProtocol.class);
        //5. 调用协议实现类
        builder.setInstance(new ClicentNameNodeImpl());
        //6. 创建服务
        RPC.Server server = builder.build();
        //7. 启动服务
        server.start();
      }
    }

其实就是通过RPC.Builder 构建一个Server对象.

Builder构建对象中包含构建Server的各种属性. 如 IP, 端口, 协议, 实现类等等. 一个builer 只能绑定一个协议和实现类.

当Builder中的各种属性填充完, 满足构建Server的条件之后, 就会构建Server对象. 并且调用Server的start方法,启动Server.

public static class Builder {
        //设置协议
        private Class<?> protocol = null;
        //设置协议的实例
        private Object instance = null;
        //设置绑定地址
        private String bindAddress = "0.0.0.0";
        //设置端口
        private int port = 0;
        //这是处理任务的hadnler数量
        private int numHandlers = 1;
        //设置读取任务的县城数量
        private int numReaders = -1;
        private int queueSizePerHandler = -1;
        private boolean verbose = false;
        private final Configuration conf;    
        private SecretManager<? extends TokenIdentifier> secretManager = null;
        private String portRangeConfig = null;
        private AlignmentContext alignmentContext = null;
        
        public Builder(Configuration conf) {
          this.conf = conf;
        }
    
    
    
    
    }

根据上面的代码,其实最核心的是创建Server服务

> **RPC.Server server = builder.build(); **

接下来,我们看看, 上面这行代码,干了什么. 通过怎样的方式构建了一个**Server服务.**

/**
         * Build the RPC Server. 
         * @throws IOException on error
         * @throws HadoopIllegalArgumentException when mandatory fields are not set
         */
        public Server build() throws IOException, HadoopIllegalArgumentException {
          if (this.conf == null) {
            throw new HadoopIllegalArgumentException("conf is not set");
          }
          if (this.protocol == null) {
            throw new HadoopIllegalArgumentException("protocol is not set");
          }
          if (this.instance == null) {
            throw new HadoopIllegalArgumentException("instance is not set");
          }
    
          //调用getProtocolEngine()获取当前RPC类配置的RpcEngine对象
          //在 NameNodeRpcServer的构造方法中已经将
          // 当前RPC类的RpcEngine对象设置为 ProtobufRpcEngine了。
          // 获取了ProtobufRpcEngine对象之后，build()方法会在
          // ProtobufRpcEngine对象上调用getServer()方法获取一个RPC Server对象的引用。
    
          return getProtocolEngine(this.protocol, this.conf).getServer(
              this.protocol, this.instance, this.bindAddress, this.port,
              this.numHandlers, this.numReaders, this.queueSizePerHandler,
              this.verbose, this.conf, this.secretManager, this.portRangeConfig,
              this.alignmentContext);
        }

看到这, 其实最主要的是getProtocolEngine方法,获取RpcEngine.  这个方法加了synchronized关键字, 所以是个同步方法.

// return the RpcEngine configured to handle a protocol
      static synchronized RpcEngine getProtocolEngine(Class<?> protocol,
          Configuration conf) {
        //从缓存中获取RpcEngine
        RpcEngine engine = PROTOCOL_ENGINES.get(protocol);
        if (engine == null) {
    
          //获取RpcEngine实现
          Class<?> impl = conf.getClass(ENGINE_PROP+"."+protocol.getName(),
                                        WritableRpcEngine.class);
    
          engine = (RpcEngine)ReflectionUtils.newInstance(impl, conf);
          PROTOCOL_ENGINES.put(protocol, engine);
        }
        return engine;
      }

RpcEngine有两种ProtobufRpcEngine和WritableRpcEngine, 默认是WritableRpcEngine(已过时). 本文代码会走默认的WritableRpcEngine.

后面会有单独的文章来解释具体的实现.

WritableRpcEngine获取Server.

/* Construct a server for a protocol implementation instance listening on a
       * port and address. */
      @Override
      public RPC.Server getServer(Class<?> protocolClass,
                          Object protocolImpl, String bindAddress, int port,
                          int numHandlers, int numReaders, int queueSizePerHandler,
                          boolean verbose, Configuration conf,
                          SecretManager<? extends TokenIdentifier> secretManager,
                          String portRangeConfig, AlignmentContext alignmentContext)
        throws IOException {
        return new Server(protocolClass, protocolImpl, conf, bindAddress, port,
            numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
            portRangeConfig, alignmentContext);
      }

ProtobufRpcEngine获取Server.

@Override
      public RPC.Server getServer(Class<?> protocol, Object protocolImpl,
          String bindAddress, int port, int numHandlers, int numReaders,
          int queueSizePerHandler, boolean verbose, Configuration conf,
          SecretManager<? extends TokenIdentifier> secretManager,
          String portRangeConfig, AlignmentContext alignmentContext)
          throws IOException {
        return new Server(protocol, protocolImpl, conf, bindAddress, port,
            numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
            portRangeConfig, alignmentContext);
      }

这两个类型的RpcEngine都会调用父类的初始化.

比如初始化  listener , handlers, responder,connectionManager 等等..

protected Server(String bindAddress, int port,
          Class<? extends Writable> rpcRequestClass, int handlerCount,
          int numReaders, int queueSizePerHandler, Configuration conf,
          String serverName, SecretManager<? extends TokenIdentifier> secretManager,
          String portRangeConfig)
        throws IOException {
        this.bindAddress = bindAddress;
        this.conf = conf;
        this.portRangeConfig = portRangeConfig;
        this.port = port;
        this.rpcRequestClass = rpcRequestClass; 
        this.handlerCount = handlerCount;
        this.socketSendBufferSize = 0;
        this.serverName = serverName;
        this.auxiliaryListenerMap = null;
        this.maxDataLength = conf.getInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH,
            CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT);
        if (queueSizePerHandler != -1) {
          this.maxQueueSize = handlerCount * queueSizePerHandler;
        } else {
          this.maxQueueSize = handlerCount * conf.getInt(
              CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
              CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);      
        }
        this.maxRespSize = conf.getInt(
            CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_KEY,
            CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_DEFAULT);
        if (numReaders != -1) {
          this.readThreads = numReaders;
        } else {
          this.readThreads = conf.getInt(
              CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
              CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
        }
        this.readerPendingConnectionQueue = conf.getInt(
            CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY,
            CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT);
    
        // Setup appropriate callqueue
        final String prefix = getQueueClassPrefix();
        this.callQueue = new CallQueueManager<Call>(getQueueClass(prefix, conf),
            getSchedulerClass(prefix, conf),
            getClientBackoffEnable(prefix, conf), maxQueueSize, prefix, conf);
    
        this.secretManager = (SecretManager<TokenIdentifier>) secretManager;
        this.authorize = 
          conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, 
                          false);
    
        // configure supported authentications
        this.enabledAuthMethods = getAuthMethods(secretManager, conf);
        this.negotiateResponse = buildNegotiateResponse(enabledAuthMethods);
        
        // Start the listener here and let it bind to the port
        listener = new Listener(port);
        // set the server port to the default listener port.
        this.port = listener.getAddress().getPort();
        connectionManager = new ConnectionManager();
        this.rpcMetrics = RpcMetrics.create(this, conf);
        this.rpcDetailedMetrics = RpcDetailedMetrics.create(this.port);
        this.tcpNoDelay = conf.getBoolean(
            CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_KEY,
            CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_DEFAULT);
    
        this.setLogSlowRPC(conf.getBoolean(
            CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC,
            CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC_DEFAULT));
    
        // Create the responder here
        responder = new Responder();
        
        if (secretManager != null || UserGroupInformation.isSecurityEnabled()) {
          SaslRpcServer.init(conf);
          saslPropsResolver = SaslPropertiesResolver.getInstance(conf);
        }
        
        this.exceptionsHandler.addTerseLoggingExceptions(StandbyException.class);
      }

看到这里就不细说了,我后面会单独拎出章节来写这两个引擎.

最后调用start 方法, 开启server服务.

/** Starts the service.  Must be called before any calls will be handled. */
      public synchronized void start() {
        responder.start();
        listener.start();
        if (auxiliaryListenerMap != null && auxiliaryListenerMap.size() > 0) {
          for (Listener newListener : auxiliaryListenerMap.values()) {
            newListener.start();
          }
        }
    
        handlers = new Handler[handlerCount];
        
        for (int i = 0; i < handlerCount; i++) {
          handlers[i] = new Handler(i);
          handlers[i].start();
        }
      }

## 2. Client 实现 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2][]

先看一下client调用server端的代码样例. 其实就是通过RPC.getProxy 方法获取server端的代理对象, 然后再通过代理对象调用具体的方法,代理对象根据方法,请求server端, 获取数据.  最终将数据返回给客户端.

/**
     * 访问RPC服务
     */
    public class Client {
      public static void main(String[] args) throws IOException {
        //1. 拿到RPC协议
        ClicentNameNodeProtocol proxy = RPC.getProxy(ClicentNameNodeProtocol.class, 1L,
            new InetSocketAddress("localhost", 7777), new Configuration());
        //2. 发送请求
        String metaData = proxy.getMetaData("/meta");
        //3. 打印元数据
        System.out.println(metaData);
      }
    }

先看一下如何获取代理对象.

/**
       * Get a protocol proxy that contains a proxy connection to a remote server
       * and a set of methods that are supported by the server
       *
       * @param protocol protocol
       * @param clientVersion client's version
       * @param addr server address
       * @param ticket security ticket
       * @param conf configuration
       * @param factory socket factory
       * @param rpcTimeout max time for each rpc; 0 means no timeout
       * @param connectionRetryPolicy retry policy
       * @param fallbackToSimpleAuth set to true or false during calls to indicate if
       *   a secure client falls back to simple auth
       * @return the proxy
       * @throws IOException if any error occurs
       */
       public static <T> ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                    long clientVersion,
                                    InetSocketAddress addr,
                                    UserGroupInformation ticket,
                                    Configuration conf,
                                    SocketFactory factory,
                                    int rpcTimeout,
                                    RetryPolicy connectionRetryPolicy,
                                    AtomicBoolean fallbackToSimpleAuth)
           throws IOException {
        if (UserGroupInformation.isSecurityEnabled()) {
          SaslRpcServer.init(conf);
        }
        return getProtocolEngine(protocol, conf).getProxy(protocol, clientVersion,
            addr, ticket, conf, factory, rpcTimeout, connectionRetryPolicy,
            fallbackToSimpleAuth, null);
      }

根据上面的方法, 先通过getProtocolEngine(protocol, conf) 这个方法, 获取到RpcEngine, 然后调用getProxy 获取对应的对象.

这个是getProxy 的接口定义, 根据RpcEngine 的不同, 实现方式也不同. 后面会有文章做讲述.

/** Construct a client-side proxy object. */
      <T> ProtocolProxy<T> getProxy(Class<T> protocol,
                      long clientVersion, InetSocketAddress addr,
                      UserGroupInformation ticket, Configuration conf,
                      SocketFactory factory, int rpcTimeout,
                      RetryPolicy connectionRetryPolicy,
                      AtomicBoolean fallbackToSimpleAuth,
                      AlignmentContext alignmentContext) throws IOException;

拿到代理对象之后, 就可以像本地一样调用里面的方法了.

参考:

Hadoop 2.X HDFS源码剖析 -- 徐鹏

[Hadoop3.2.1 _ HDFS _ _ RPC_ _ proto]: https://mp.csdn.net/console/editor/html/105586527
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20230521/e3d6a39ff35249839fd129c2ee5141b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20230521/af5961bfa32441adb064221665d59676.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20230521/fcd23bf2b4114e3dba47d786269e512b.png