Hadoop中RPC机制

Hadoop中RPC机制

RPC(Remote Procedure Call Protocol)远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。Hadoop底层的交互都是通过rpc进行的。例如：datanode和namenode 、tasktracker和jobtracker、secondary namenode和namenode之间的通信都是通过rpc实现的。下面是rpc交互过程图： 

1.客服端调用的总过程：

Hadoop的RPC客户端代码其实就一个类：org.apache.hadoop.ipc.Client。这个类使用Java的动态代理技术，生成服务器的业务接口的代理，通过socket将调用的业务方法和参数传送到服务器端，并且等待服务器端的响应。

客户端调用的序列图如下:

 

 例如：TaskTracker请求与JobTracker的通信：

TaskTracker通过:

      this.jobClient = (InterTrackerProtocol)

          RPC.waitForProxy(InterTrackerProtocol.class, InterTrackerProtocol.versionID,

                       jobTrackAddr, this.fConf);

中的InterTrackerProtocol 去和JobTracker通信。

在RPC中通过:

       VersionedProtocol proxy =

            (VersionedProtocol) Proxy.newProxyInstance(

            protocol.getClassLoader(), new Class[] { protocol },

            new Invoker(addr, ticket, conf, factory));

产生一个动态代理完成JobTracker和TaskTracker之间的心跳交流。

2.客服端向服务器发送连接

    客户端（C）要发起对服务端（S）方法的调用主要通过：

    public Writable call(Writable param, InetSocketAddress addr,

                       Class<?> protocol, UserGroupInformation ticket) 

                       throws InterruptedException, IOException 实现

2.1首先在该方法调用中：

    Call call = new Call(param); //将param转换成Call对象  其实就是将Invocation（用来序列化和反序列化RPC客户端的调用信息，包括方法名和参数信息）转化为Call 。

2.2 客服端创建一个通向服务端的连接connection，Connection connection = getConnection(addr, protocol, ticket, call);然后将此次调用放入CallList里，这样客户端就可以同时发生很多调用，每个调用用ID来识别。

(1) 根据RPC服务端的地址和接口从连接池中获取一个，如果取到Connection则直接返回。

(2) 否则新建一个Connection，并将它放入到连接池中

(3) 然后通过SocketFactory创建一个Socket，并建立到RPC服务端的连接，如果连接不成功，则重试。

(4) 创建和关联输入和输出流对象。

2.3 发送调用参数connection.sendParam(call)。调用参数是Client的调用方（比如NameNode，DataNode等）指定的，一般就是一个Invocation对象，里面包含要调用的方法和参数。

2.4等待调用结果.Client.Connection是个线程类，启动了之后唯一做的事情就是等待调用结果。

    synchronized (call) {   

          while (!call.done){   //done就是服务器端返回该call的结果，判断该call是否处理

          ……………………           

          ………………

       }

    }

3.服务器端对客服端的call请求处理

对于服务器端，其有一个方法start指定了启动服务器开始监听，这个start被四个类调用，分别是TaskTracker.initialize，Namenode.initialize,Jobtracker.offerService,Datanode.startDatanode，显然，任何两者之间的通信都是考这个client-server模型实现的。

3.1 server start后，干了三件事，就是启动三个线程

1.启动listener，监听客户端Call

2.启动response，随时准备将处理结果发回client

3.启动10个handler，处理具体的请求。

3.2上面三个线程的具体工作过程

3.2.1. Listener线程

该线程负责监听客户端请求以及数据的接收，然后将接收到的数据组成一个Call实例，放到请求队列里面。具体做法如下：

1) Listener线程循环等待RPC客户端的发送数据过来

2) 当有数据可以接收时，调用Connection的readAndProcess（）方法。readAndProces（）又调用processData()方法

3) Connection边接收边对数据进行处理，如果接收到一个完整的Call包，则构建一个Call对象。readAndProcess（）调用processData()方法把Call对象加入到Call队列的，并将这个Call对象PUSH到Call队列中，由Handler线程来处理Call队列中的所有Call。

3.2.2.Handler线程

该线程负责从请求队列中取出调用请求，通过调用抽象方法

1) Handler线程循环监听Call队列，如果Call队列为空，则进入wait状态，否则按FIFO规则从Call队列取出Call

2) 将Call交给RPC.Server处理（调用RPC.Server的public Writable call(Class<?> protocol, Writable param, long receivedTime)

    throws IOException ），因为RPC对Server的一些功能进行了实现

3) 借助java里的反射机制，完成对目标方法的调用

4) 返回响应。由于响应需要通过SOCKET返回给RPC客户端，所以响应的类型必须是Writable。

3.2.2. Response线程   

 Response也监视responselist，如果responselist中某个call需要将结果写入客户端就写出，当某个call的结果被发送完毕，从responselist中删除该call对象。

注意：handler完成call之后就开始向客户端写call结果，但是结果可能太多，无法通过一次性发送完毕，而发送之后还要等待client接受完毕才能再发，如果现在handler在那里等待客户端接受完毕，然后再发，效率不高。解决办法是handler处理完毕之后，只向client发送一次处理结果。如果这一次将处理结果发送完毕，接下来就没有response的事情了，如果没有发送完毕，接下来response负责将剩下的处理结果发送给客户端。这样handler的并发量会大一些。

服务器实现中大量利用监视队列，比如handler就直观坚持calllist，一旦发现数据就开始处理，而response就监视responselist，发现数据需要发送就开始发送。

参考：

1.http://bbs.hadoopor.com/thread-329-1-2.html

2.http://jackosn-liao.iteye.com/blog/851914 

3. http://blog.csdn.net/wuixiaobao/article/details/6549781

4.http://bbs.hadoopor.com/thread-329-1-2.html