当前版本Hadoop RPC

电玩女神 2022-06-06 00:20 185阅读 0赞

**什么是RPC?**

1、RPC(Remote Procedure Call）远程过程调用，它允许一台计算机程序远程调用另外一台计算机的子程序，而**不用去关心底层的网络通信细节**，对我们来说是透明的。经常用于分布式网络通信中。

2、Hadoop的进程间交互都是通过RPC来进行的，比如Namenode与Datanode之间，Jobtracker与Tasktracker之间等。

RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中， RPC跨越了传输层和应用层。 RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用**客户机/服务器**模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。

首先，**客户机**调用进程**发送**一个有进程参数的调用信息到**服务进程**，然后等待应答信息，在服务器端，进程**保持睡眠状态**直到**调用信息的到达**为止。当一个调用信息到达，**服务器获得进程参数，计算结果，发送答复信息给client，**然后**等待下一个调用信息**，最后，客户端调用进程**接收答复信息，获得进程结果**，然后**调用执行继续进行。**

**RPC特点**

1、透明性：远程调用其他机器上的程序，对用户来说就像是调用本地方法一样。

2、高性能：RPC server能够并发处理多个来自Client的请求（请求队列）。3、可控性：jdk中已经提供了一个RPC框架–RMI，但是该RPC框架过于重量级并且可控之处比较少，所以Hadoop RPC实现了自定义的RPC框架。

**Hadoop RPC通信**

1、序列化层：Client与Server端 通信传递的信息采用了Hadoop里提供的序列化类或自定义Writable类型。

2、函数调用层：Hadoop RPC通过**动态代理**以及**Java反射机制**实现函数调用。

3、网络传输层：Hadoop RPC采用了基于TCP/IP的socket机制。

4、服务器端框架层：RPC Server利用**Java NIO**以及采用了**事件驱动的I/O模型**，提高**RPC Server的并发处理能力**。

Hadoop的整个体系结构就是构建在RPC之上(org.apache.hadoop.ipc)。

**Hadoop RPC设计技术**

1、动态代理

2、反射3、序列化4、非阻塞的异步IO（NIO）

**动态代理**

1、动态代理可以提供对另一个对象的访问，同时隐藏实际对象的具体事实，代理对象对客户隐藏了实际对象。

2、动态代理可以对请求进行其他的一些处理，在不允许直接访问某些类，或需要对访问做一些特殊处理等，这时候可以考虑使用代理。3）目前Java开发包中提供了对动态代理的支持，但现在只支持对接口的实现。**相关的类与接口：**`java.lang.reflect.Proxy--类 java.lang.reflect.InvocationHandler--接口`

**动态代理创建对象过程：**

`InvocationHandler handler = new InvocationHandlerImpl(...) Proxy.newInstance(...)`

具体实现可参考如下：

[![Hadoop RPC远程过程调用源码解析及实例][Hadoop RPC]][Hadoop RPC]

**根据上图查看hadoop2.6.0源码**

**几个重要的协议**

**ClientProtocol**是客户端(FileSystem)与NameNode通信的接口。

**DatanodeProtocol**是DataNode与NameNode通信的接口**NamenodeProtocol**是SecondaryNameNode与NameNode通信的接口。**DFSClient**是直接调用NameNode接口的对象。用户代码是通过DistributedFileSystem调用DFSClient对象，才能与NameNode打交道。

![wKiom1dFxy7zXLBrAACsqfKTwcY443.jpg-wh_600x-s_2423081374.jpg][]

![wKiom1dFxy7yrIL_AADYRXd95XU074.jpg-s_2838660135.jpg][]

**模拟Hadoop RPC通信**

package MyRPC; 
       import org.apache.hadoop.io.Text; 
       import org.apache.hadoop.ipc.VersionedProtocol; 
        
       public interface MyRPCProtocal extends VersionedProtocol{ 
           public static long versionID = 23234l;//很重要很重要，搞了一下午才解决掉。 
           public Text test(Text t); 
       }

package MyRPC; 
        
       import java.io.IOException; 
       import org.apache.hadoop.conf.Configuration; 
       import org.apache.hadoop.io.Text; 
       import org.apache.hadoop.ipc.ProtocolSignature; 
       import org.apache.hadoop.ipc.RPC; 
       import org.apache.hadoop.ipc.RPC.Server; 
        
       public class RPCServer implements MyRPCProtocal{     
           Server server = null; 
           public RPCServer() throws IOException, InterruptedException{ 
               //server = RPC.getServer(this,"localhost",8888,new Configuration()); 
               //相对于以前的版本有略微的改动 
               RPC.Builder ins = new RPC.Builder(new Configuration()); 
               ins.setInstance(this); 
               ins.setBindAddress("localhost"); 
               ins.setPort(9999); 
               ins.setProtocol(MyRPCProtocal.class); 
               //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class); 
               server = ins.build();//获得一个server实例 
               server.start(); 
               server.join();   
           } 
      
     
      
      
       
     
      
     
      
      
       
     
      
     
      
        
           public static void main(String[] args) throws IOException, InterruptedException { 
               new RPCServer(); 
           } 
        
           @Override 
           public long getProtocolVersion(String protocol, long clientVersion) 
                   throws IOException { 
               return MyRPCProtocal.versionID; 
           } 
        
           @Override 
           public ProtocolSignature getProtocolSignature(String protocol, 
                   long clientVersion, int clientMethodsHash) throws IOException {      
               return new ProtocolSignature(); 
           } 
        
           @Override 
           public Text test(Text t) { 
               if(t.toString().equals("RPC")){ 
                   return new Text("ok"); 
               } 
               return new Text("false"); 
           } 
       } 
       package MyRPC; 
        
       import java.net.InetSocketAddress; 
        
       import org.apache.hadoop.conf.Configuration; 
       import org.apache.hadoop.io.Text; 
       import org.apache.hadoop.ipc.RPC; 
        
       public class RPCClient { 
        
           private MyRPCProtocal protocal; 
        
           public RPCClient() throws Exception{ 
               InetSocketAddress address = new InetSocketAddress("localhost",9999); 
        
               protocal = (MyRPCProtocal)RPC.waitForProxy 
                       (MyRPCProtocal.class,MyRPCProtocal.versionID, address, new Configuration()); 
               //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class); 
           } 
        
           public void call(String s){ 
               final Text string = protocal.test(new Text(s)); 
               System.out.println(string.toString()); 
           } 
        
           public static void main(String[] args) throws Exception { 
               RPCClient client = new RPCClient(); 
               client.call("RPC"); 
           } 
       } 
      
     
      
      
       
     
      
     
      
      
       
     
      
     
      
      
       
      
       3.3.4　Hadoop RPC使用方法
      
       Hadoop RPC对外主要提供了两种接口（见类org.apache.hadoop.ipc.RPC），分别是：
      
       public static <T> ProtocolProxy <T> getProxy/waitForProxy(…)：构造一个客户端代理对象（该对象实现了某个协议），用于向服务器发送RPC请求。
      
       public static Server RPC.Builder (Configuration).build()：为某个协议（实际上是Java接口）实例构造一个服务器对象，用于处理客户端发送的请求。
      
       通常而言，使用Hadoop RPC可分为以下4个步骤。
      
       1. 定义RPC协议
      
       RPC协议是客户端和服务器端之间的通信接口，它定义了服务器端对外提供的服务接口。如下所示，我们定义一个ClientProtocol通信接口，声明了echo()和add()两个方法。需要注意的是，Hadoop中所有自定义RPC接口都需要继承VersionedProtocol接口，它描述了协议的版本信息。 
      
       
       
        
         interface ClientProtocol extends org.apache.hadoop.ipc.VersionedProtocol {  
         //版本号，默认情况下，不同版本号的RPC Client和Server之间不能相互通信  
             public static final long versionID = 1L;  
             String echo(String value) throws IOException;  
             int add(int v1, int v2) throws IOException;  
           } 
        
      
       2. 实现RPC协议
      
       Hadoop RPC协议通常是一个Java接口，用户需要实现该接口。对ClientProtocol接口进行简单的实现如下所示： 
      
       
       
        
         public static class ClientProtocolImpl implements ClientProtocol {  
             //重载的方法，用于获取自定义的协议版本号，  
             public long getProtocolVersion(String protocol, long clientVersion) {  
               return ClientProtocol.versionID;  
              }  
             //重载的方法，用于获取协议签名  
             public ProtocolSignature getProtocolSignature(String protocol, long clientVersion,  
             inthashcode) {  
               return new ProtocolSignature(ClientProtocol.versionID, null);  
             }  
             public String echo(String value) throws IOException {  
               return value;  
              }  
             public int add(int v1, int v2) throws IOException {  
               return v1 + v2;  
              }  
           } 
        
      
       3. 构造并启动RPC Server
      
       直接使用静态类Builder构造一个RPC Server，并调用函数start()启动该Server： 
      
       
       
        
         Server server = new RPC.Builder(conf).setProtocol(ClientProtocol.class)  
            .setInstance(new ClientProtocolImpl()).setBindAddress(ADDRESS).setPort(0)  
                                     .setNumHandlers(5).build();  
         server.start(); 
        
      
       其中，BindAddress（由函数setBindAddress设置）和Port（由函数setPort设置，0表示由系统随机选择一个端口号）分别表示服务器的host和监听端口号，而NnumHandlers（由函数setNumHandlers设置）表示服务器端处理请求的线程数目。到此为止，服务器处理监听状态，等待客户端请求到达。
      
       4. 构造RPC Client并发送RPC请求
      
       使用静态方法getProxy构造客户端代理对象，直接通过代理对象调用远程端的方法，具体如下所示： 
      
       
       
        
         proxy = (ClientProtocol)RPC.getProxy(  
                  ClientProtocol.class, ClientProtocol.versionID, addr, conf);  
         int result = proxy.add(5, 6);  
         String echoResult = proxy.echo("result"); 
        
      
       经过以上四步，我们便利用Hadoop RPC搭建了一个非常高效的客户机–服务器网络模型。接下来，我们将深入到Hadoop RPC内部，剖析它的设计原理及技巧。

[Hadoop RPC]: /images/20220606/f0010242d2d34691869bea6ce0ea4607.png
[wKiom1dFxy7zXLBrAACsqfKTwcY443.jpg-wh_600x-s_2423081374.jpg]: /images/20220606/05574db5e0504e6cad5829c383c31e98.png
[wKiom1dFxy7yrIL_AADYRXd95XU074.jpg-s_2838660135.jpg]: /images/20220606/143e7db0cd644fd2b5a96dc4b9e1823d.png