【公司要求】- RPC学习（一）

HADOOP-IPC(这里说的是1.0.4版本) 是轻量级RPC，在hadoop中主要用于2方面

　　1、TaskTracker和JobTracker 通讯。

　　2、NameNode和DataNode通讯。

　　3、DataNode和DataNode通讯。

作者的意图是尽量暴露细节参数给用户从而达到增加灵活性（比如连接超时设置、并发业务处理器数量、队列长度设置等）。核心代码3000行左右。采用基于java-NIO非阻塞方式，设计中间有一些细节是很值得学习的。下图展示了IPC总体设计方式（其实是网路服务器架构方式，缺少业务处理模handler） hadoop-IPC设计

一些细节：

　　　　　　1、hadoop更新版本迅速，所以IPC在内部通讯的时候都实现了VersionProtocal接口，不同的版本的Server和Client是不能通讯的。

　　　　　　2、新的版本（2.0.0）中加入了PRCEngine用于扩展第三方RPC框架的接入，增大灵活性。

　　　　　　3、Client由于是异步调用所以无法保证消息顺序，因此Client把ConnectionId类作为Key放到了本地缓存中。返回一个remove一个。

　　　　　　4、Client做了链接复用。而且用户可以回收连接时间间隔（当连接空闲的时候）。

　　　　　　5、Server端Responder是一个独立的线程服务，作用是防止RPC调用时间过长而阻塞服务处理器（Handler是共享资源，要提高并发就要尽量减少公共资源的使用时间）。因此如果Server中只有一个请求业务时（表明服务端压力不大），Handler直接打包返回。如果大于1，则Handler的持有线程（业务线程）。把请求转交给Responder线程处理，同事Register通道的OP_WRITE写事件。后续和Client交互又Responder处理。

　　　　　　6、Listener和Reader分离，每部分都利用了NIO的Reactor模式。Listener处理网络监听连接事件，Reader由Listener启动负责打包成Server可以接受的对象放到业务处理器要处理的队列中。

使用DEMO：

　　我们在使用Hadoop——IPC做我们自己的分布式系统的基础框架也是比较方便的，需要引用Hadoop-core-x.x.x.jar,其实还是建议抽取出来加以改进更好的支持业务。最后下面引用某教材。

0、边界业务接口和实现

1 public interface IPCQueryStatus extends VersionedProtocol {
2     IPCFileStatus getFileStatus(String filename);
3 }

 1 public class IPCQueryStatusImpl implements IPCQueryStatus {    
 2     protected IPCQueryStatusImpl() {
 3     }
 4 
 5     @Override
 6     public IPCFileStatus getFileStatus(String filename) {
 7         IPCFileStatus status=new IPCFileStatus(filename);
 8         System.out.println("Method getFileStatus Called, return: "+status);
 9         return status;
10     }
11 
12     @Override
13     public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
14         System.out.println("protocol: "+protocol);
15         System.out.println("clientVersion: "+clientVersion);
16         return IPCQueryServer.IPC_VER;
17     }
18 }

 1 public class IPCFileStatus implements Writable {    
 2     private String filename;
 3     private long time;
 4     
 5     static {   // register IPCFileStatus
 6         WritableFactories.setFactory
 7             ( IPCFileStatus.class,
 8               new WritableFactory() {
 9                   public Writable newInstance() { return new IPCFileStatus(); } } );
10     }
11     public IPCFileStatus() {        
12     }
13     
14     public IPCFileStatus(String filename) {
15         this.filename=filename;
16         this.time=(new Date()).getTime();
17     }
18 
19     public String getFilename() {
20         return filename;
21     }
22 }

1、开启server

1 IPCQueryStatusImpl queryService=new IPCQueryStatusImpl();
2 Server server = RPC.getServer(queryService, "0.0.0.0", IPC_PORT, 1, true,new Configuration());
3 server.start();

2、开启client

InetSocketAddress addr=new InetSocketAddress("localhost", IPCQueryServer.IPC_PORT);
IPCQueryStatus query=(IPCQueryStatus) RPC.getProxy(IPCQueryStatus.class, IPCQueryServer.IPC_VER, addr,  new Configuration());
IPCFileStatus status=query.getFileStatus("/tmp/testIPC");

下期写Hadoop的具体实现细节和基础知识回顾。