RPC概述

RPC（Remote Procedure Call）即远程过程调用，是一种通过网络从远程计算机程序上请求服务的协议。RPC允许本地程序像调用本地方法一样调用远程计算机上的应用程序，其使用常见的网络传输协议（如TCP或UDP）传递RPC请求以及相应信息，使得分布式程序的开发更加容易。Hadoop作为分布式存储系统， 各个节点之间的通信和交互是必不可少的， 所以需要实现一套节点间的通信交互机制

Hadoop实现了一套自己的RPC框架。采用了JavaNIO、Java动态代理以及protobuf等基础技术

RPC采用客户端／服务器模式，请求程序就是一个客户端，而服务提供程序就是一个服务器。客户端首先会发送一个有参数的调用请求到服务器，然后等待服务器发回响应信息。在服务器端，服务提供程序会保持睡眠状态直到有调用请求到达为止。当一个调用请求到达后，服务提供程序会执行调用请求，计算结果，向客户端发送响应信息，然后等待下一个调用请求。最后，客户端成功地接收服务器发回的响应信息，一个远程调用结束
image.png

  • client functions

    • 请求程序，会像调用本地方法一样调用客户端stub程序（如图中①），然后接受stub程序的响应信息（如图中⑩）

  • client stub

    • 客户端stub程序，表现得就像本地程序一样，但底层却会调用请求和参数序列化并通过通信模块发送给服务器（如图中②）；客户端stub程序也会等待服务器的响应信息（如图中⑨），将响应信息反序列化并返回给请求程序（如图中⑩）

      序列化：将对象写入IO流，目的是使对象可以脱离程序的运行而独立存在

  • sockets

    • 网络通信模块，用于传输RPC请求和响应（如图中的③⑧），可以基于TCP或UDP协议

  • server stub

    • 服务端stub程序，会接收客户端发送的请求和参数（如图中④）并反序列化，根据调用信息触发对应的服务程序（如图中⑤），然后将服务程序的响应信息（如图⑥），并序列化并发回给客户端（如图中⑦）

  • server functions

    • 服务程序，会接收服务端stub程序的调用请求（如图中⑤），执行对应的逻辑并返回执行结果（如图中⑥）

Hadoop RPC使用

Hadoop RPC框架主要由三个类组成： RPC、 Client和Server类

  • RPC类用于对外提供一个使用Hadoop RPC框架的接口

  • Client类用于实现RPC客户端功能

  • Server类则用于实现RPC服务器端功能

首先写一个demo来测试一下效果，然后再说RPC理论。这个demo采用的是hadoop默认的RPC Engine : WritableRpcEngine。采用proto协议在定义接口协议&实现方面会不一样，后面会有这两个rpc engine的调度实例

定义接口协议

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

实现接口协议

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

创建Server服务，并注册协议，启动RPC服务

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

创建Client服务，请求数据接口

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

Hadoop RPC原理

定义接口协议

根据业务需要定义接口协议

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

实现接口协议

根据接口创建一个实现类，用的时候注册到Server服务中即可

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

image.png

RPC Server 处理流程

Server架构图

image.png

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将剩余响应发回客户端

Server组件代码

Server服务是用RPC.Builder类中的build()方法构建的

  1. /**
  2.  * 构建Server代码
  3.  */
  4. public class Server {
  5.   public static void main(String[] args) throws IOException {
  6.     //1. 构建RPC框架
  7.     RPC.Builder builder = new RPC.Builder(new Configuration());
  8.     //2. 绑定地址
  9.     builder.setBindAddress("localhost");
  10.     //3. 绑定端口
  11.     builder.setPort(7777);
  12.     //4. 绑定协议
  13.     builder.setProtocol(ClicentNameNodeProtocol.class);
  14.     //5. 调用协议实现类
  15.     builder.setInstance(new ClicentNameNodeImpl());
  16.     //6. 创建服务
  17.     RPC.Server server = builder.build();
  18.     //7. 启动服务
  19.     server.start();
  20.   }
  21. }

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

Builder构建对象中包含构建Server的各种属性，一个builder只能绑定一个协议和实现类

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

  1. public static class Builder {
  2.     //设置协议
  3.     private Class<?> protocol = null;
  4.     //设置协议的实例
  5.     private Object instance = null;
  6.     //设置绑定地址
  7.     private String bindAddress = "0.0.0.0";
  8.     //设置端口
  9.     private int port = 0;
  10.     //这是处理任务的hadnler数量
  11.     private int numHandlers = 1;
  12.     //设置读取任务的线程数量
  13.     private int numReaders = -1;
  15.     private int queueSizePerHandler = -1;
  16.     private boolean verbose = false;
  17.     private final Configuration conf;    
  18.     private SecretManager<? extends TokenIdentifier> secretManager = null;
  19.     private String portRangeConfig = null;
  20.     private AlignmentContext alignmentContext = null;
  22.     public Builder(Configuration conf) {
  23.       this.conf = conf;
  24.     } 
  25. }

最核心的代码是创建Server服务

  1. RPC.Server server = builder.build();

我们来分析一下这行代码做了什么，通过怎样的方式来构建了一个Server服务

  1.    /**
  2.      * Build the RPC Server. 
  3.      * @throws IOException on error
  4.      * @throws HadoopIllegalArgumentException when mandatory fields are not set
  5.      */
  6.     public Server build() throws IOException, HadoopIllegalArgumentException {
  7.       if (this.conf == null) {
  8.         throw new HadoopIllegalArgumentException("conf is not set");
  9.       }
  10.       if (this.protocol == null) {
  11.         throw new HadoopIllegalArgumentException("protocol is not set");
  12.       }
  13.       if (this.instance == null) {
  14.         throw new HadoopIllegalArgumentException("instance is not set");
  15.       }
  17.       // 调用getProtocolEngine()获取当前RPC类配置的RpcEngine对象
  18.       // 在NameNodeRpcServer的构造方法中已将当前RPC类的RpcEngine对象设置为ProtobufRpcEngine
  19.       // 获取了ProtobufRpcEngine对象之后，build()方法会在
  20.       // ProtobufRpcEngine对象上调用getServer()方法来获取一个RPC Server对象的引用
  22.       return getProtocolEngine(this.protocol, this.conf).getServer(
  23.           this.protocol, this.instance, this.bindAddress, this.port,
  24.           this.numHandlers, this.numReaders, this.queueSizePerHandler,
  25.           this.verbose, this.conf, this.secretManager, this.portRangeConfig,
  26.           this.alignmentContext);
  27.     }

其实最主要的是**getProtocolEngine**方法，获取RpcEngine，此方法加了Synchronized关键字，是同步方法

  1. // return the RpcEngine configured to handle a protocol
  2.   static synchronized RpcEngine getProtocolEngine(Class<?> protocol,
  3.       Configuration conf) {
  4.     // 从缓存中获取RpcEngine
  5.     RpcEngine engine = PROTOCOL_ENGINES.get(protocol);
  6.     if (engine == null) {
  8.     //获取RpcEngine实现
  9.           Class<?> impl = conf.getClass(ENGINE_PROP+"."+protocol.getName(),
  10.                                         WritableRpcEngine.class);
  12.           engine = (RpcEngine)ReflectionUtils.newInstance(impl, conf);
  13.           PROTOCOL_ENGINES.put(protocol, engine);
  14.     }
  15.     return engine;
  16.   }

RpcEngine有两种ProtobufRpcEngine和WritableRpcEngine，默认是WritableRpcEngine

WritableRpcEngine获取Server

  1.  /* Construct a server for a protocol implementation instance listening on a
  2.   * port and address. */
  3.   @Override
  4.   public RPC.Server getServer(Class<?> protocolClass,
  5.                       Object protocolImpl, String bindAddress, int port,
  6.                       int numHandlers, int numReaders, int queueSizePerHandler,
  7.                       boolean verbose, Configuration conf,
  8.                       SecretManager<? extends TokenIdentifier> secretManager,
  9.                       String portRangeConfig, AlignmentContext alignmentContext)
  10.     throws IOException {
  11.     return new Server(protocolClass, protocolImpl, conf, bindAddress, port,
  12.         numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
  13.         portRangeConfig, alignmentContext);
  14.   }

ProtobufRpcEngine获取Server

  1.  @Override
  2.   public RPC.Server getServer(Class<?> protocol, 
  3.                       Object protocolImpl,String bindAddress, int port, 
  4.                       int numHandlers, int numReaders, int queueSizePerHandler, 
  5.                       boolean verbose, Configuration conf,
  6.                         SecretManager<? extends TokenIdentifier> secretManager,
  7.                         String portRangeConfig, AlignmentContext alignmentContext)
  8.     throws IOException {
  9.     return new Server(protocol, protocolImpl, conf, bindAddress, port,
  10.         numHandlers, numReaders, queueSizePerHandler, verbose, secretManager,
  11.         portRangeConfig, alignmentContext);
  12.   }

这两个类型的RpcEngine都会调用父类的初始化，比如初始化listener、handlers、responder、connectionManager等等

  1. protected Server(String bindAddress, int port,
  2.               Class<? extends Writable> rpcRequestClass, int handlerCount,
  3.               int numReaders, int queueSizePerHandler, Configuration conf,
  4.               String serverName, SecretManager<? extends TokenIdentifier> secretManager,
  5.               String portRangeConfig)
  6.     throws IOException {
  7.     this.bindAddress = bindAddress;
  8.     this.conf = conf;
  9.     this.portRangeConfig = portRangeConfig;
  10.     this.port = port;
  11.     this.rpcRequestClass = rpcRequestClass; 
  12.     this.handlerCount = handlerCount;
  13.     this.socketSendBufferSize = 0;
  14.     this.serverName = serverName;
  15.     this.auxiliaryListenerMap = null;
  16.     this.maxDataLength = conf.getInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH,
  17.         CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT);
  18.     if (queueSizePerHandler != -1) {
  19.       this.maxQueueSize = handlerCount * queueSizePerHandler;
  20.     } else {
  21.       this.maxQueueSize = handlerCount * conf.getInt(
  22.           CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
  23.           CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);      
  24.     }
  25.     this.maxRespSize = conf.getInt(
  26.         CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_KEY,
  27.         CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_DEFAULT);
  28.     if (numReaders != -1) {
  29.       this.readThreads = numReaders;
  30.     } else {
  31.       this.readThreads = conf.getInt(
  32.           CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
  33.           CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
  34.     }
  35.     this.readerPendingConnectionQueue = conf.getInt(
  36.         CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY,
  37.         CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT);
  39.     // Setup appropriate callqueue
  40.     final String prefix = getQueueClassPrefix();
  41.     this.callQueue = new CallQueueManager<Call>(getQueueClass(prefix, conf),
  42.         getSchedulerClass(prefix, conf),
  43.         getClientBackoffEnable(prefix, conf), maxQueueSize, prefix, conf);
  45.     this.secretManager = (SecretManager<TokenIdentifier>) secretManager;
  46.     this.authorize = 
  47.       conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, 
  48.                       false);
  50.     // configure supported authentications
  51.     this.enabledAuthMethods = getAuthMethods(secretManager, conf);
  52.     this.negotiateResponse = buildNegotiateResponse(enabledAuthMethods);
  54.     // Start the listener here and let it bind to the port
  55.     listener = new Listener(port);
  56.     // set the server port to the default listener port.
  57.     this.port = listener.getAddress().getPort();
  58.     connectionManager = new ConnectionManager();
  59.     this.rpcMetrics = RpcMetrics.create(this, conf);
  60.     this.rpcDetailedMetrics = RpcDetailedMetrics.create(this.port);
  61.     this.tcpNoDelay = conf.getBoolean(
  62.         CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_KEY,
  63.         CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_DEFAULT);
  65.     this.setLogSlowRPC(conf.getBoolean(
  66.         CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC,
  67.         CommonConfigurationKeysPublic.IPC_SERVER_LOG_SLOW_RPC_DEFAULT));
  69.     // Create the responder here
  70.     responder = new Responder();
  72.     if (secretManager != null || UserGroupInformation.isSecurityEnabled()) {
  73.       SaslRpcServer.init(conf);
  74.       saslPropsResolver = SaslPropertiesResolver.getInstance(conf);
  75.     }
  77.     this.exceptionsHandler.addTerseLoggingExceptions(StandbyException.class);
  78.   }

最后调用start方法，开启Server服务

  1.   /** Starts the service.  Must be called before any calls will be handled. */
  2.   public synchronized void start() {
  3.     responder.start();
  4.     listener.start();
  5.     if (auxiliaryListenerMap != null && auxiliaryListenerMap.size() > 0) {
  6.         for (Listener newListener : auxiliaryListenerMap.values()) {
  7.         newListener.start();
  8.         }
  9.     }
  11.     handlers = new Handler[handlerCount];
  13.     for (int i = 0; i < handlerCount; i++) {
  14.           handlers[i] = new Handler(i);
  15.           handlers[i].start();
  16.         }
  17.   }

Client实现

image.png

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

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

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

  1.   /**
  2.    * Get a protocol proxy that contains a proxy connection to a remote server
  3.    * and a set of methods that are supported by the server
  4.    *
  5.    * @param protocol protocol
  6.    * @param clientVersion client's version
  7.    * @param addr server address
  8.    * @param ticket security ticket
  9.    * @param conf configuration
  10.    * @param factory socket factory
  11.    * @param rpcTimeout max time for each rpc; 0 means no timeout
  12.    * @param connectionRetryPolicy retry policy
  13.    * @param fallbackToSimpleAuth set to true or false during calls to indicate if
  14.    *   a secure client falls back to simple auth
  15.    * @return the proxy
  16.    * @throws IOException if any error occurs
  17.    */
  18.    public static <T> ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
  19.                                 long clientVersion,
  20.                                 InetSocketAddress addr,
  21.                                 UserGroupInformation ticket,
  22.                                 Configuration conf,
  23.                                 SocketFactory factory,
  24.                                 int rpcTimeout,
  25.                                 RetryPolicy connectionRetryPolicy,
  26.                                 AtomicBoolean fallbackToSimpleAuth)
  27.     throws IOException {
  28.     if (UserGroupInformation.isSecurityEnabled()) {
  29.           SaslRpcServer.init(conf);
  30.     }
  31.     return getProtocolEngine(protocol, conf).getProxy(protocol, clientVersion,
  32.         addr, ticket, conf, factory, rpcTimeout, connectionRetryPolicy,
  33.         fallbackToSimpleAuth, null);
  34.       }

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

这个是getProxy 的接口定义，根据RpcEngine 的不同，实现方式也不同

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

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