什么是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(...)

　　具体实现可参考如下：

　　根据上图查看hadoop2.6.0源码

　　Client

　　Server

　　RPC

　　几个重要的协议

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

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

　　模拟Hadoop RPC通信

1. package MyRPC; 


2. import org.apache.hadoop.io.Text; 


3. import org.apache.hadoop.ipc.VersionedProtocol; 


4.  


5. public interface MyRPCProtocal extends VersionedProtocol{ 


6.     public static long versionID = 23234l;//很重要很重要，搞了一下午才解决掉。 


7.     public Text test(Text t); 


8. } 

1. package MyRPC; 


2.  


3. import java.io.IOException; 


4. import org.apache.hadoop.conf.Configuration; 


5. import org.apache.hadoop.io.Text; 


6. import org.apache.hadoop.ipc.ProtocolSignature; 


7. import org.apache.hadoop.ipc.RPC; 


8. import org.apache.hadoop.ipc.RPC.Server; 


9.  


10. public class RPCServer implements MyRPCProtocal{     


11.     Server server = null; 


12.     public RPCServer() throws IOException, InterruptedException{ 


13.         //server = RPC.getServer(this,'localhost',8888,new Configuration()); 


14.         //相对于以前的版本有略微的改动 


15.         RPC.Builder ins = new RPC.Builder(new Configuration()); 


16.         ins.setInstance(this); 


17.         ins.setBindAddress('localhost'); 


18.         ins.setPort(9999); 


19.         ins.setProtocol(MyRPCProtocal.class); 


20.         //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class); 


21.         server = ins.build();//获得一个server实例 


22.         server.start(); 


23.         server.join();   


24.     } 


25.  


26.     public static void main(String[] args) throws IOException, InterruptedException { 


27.         new RPCServer(); 


28.     } 


29.  


30.     @Override 


31.     public long getProtocolVersion(String protocol, long clientVersion) 


32.             throws IOException { 


33.         return MyRPCProtocal.versionID; 


34.     } 


35.  


36.     @Override 


37.     public ProtocolSignature getProtocolSignature(String protocol, 


38.             long clientVersion, int clientMethodsHash) throws IOException {     


39.         return new ProtocolSignature(); 


40.     } 


41.  


42.     @Override 


43.     public Text test(Text t) { 


44.         if(t.toString().equals('RPC')){ 


45.             return new Text('ok'); 


46.         } 


47.         return new Text('false'); 


48.     } 


49. } 


50. package MyRPC; 


51.  


52. import java.net.InetSocketAddress; 


53.  


54. import org.apache.hadoop.conf.Configuration; 


55. import org.apache.hadoop.io.Text; 


56. import org.apache.hadoop.ipc.RPC; 


57.  


58. public class RPCClient { 


59.  


60.     private MyRPCProtocal protocal; 


61.  


62.     public RPCClient() throws Exception{ 


63.         InetSocketAddress address = new InetSocketAddress('localhost',9999); 


64.  


65.         protocal = (MyRPCProtocal)RPC.waitForProxy 


66.                 (MyRPCProtocal.class,MyRPCProtocal.versionID, address, new Configuration()); 


67.         //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class); 


68.     } 


69.  


70.     public void call(String s){ 


71.         final Text string = protocal.test(new Text(s)); 


72.         System.out.println(string.toString()); 


73.     } 


74.  


75.     public static void main(String[] args) throws Exception { 


76.         RPCClient client = new RPCClient(); 


77.         client.call('RPC'); 


78.     } 


79. }