【Hadoop代码笔记】通过JobClient对Jobtracker的调用详细了解Hadoop RPC

zenmbu

　　Hadoop的各个服务间，客户端和服务间的交互采用RPC方式。关于这种机制介绍的资源很多，也不难理解，这里不做背景介绍。只是尝试从Jobclient向JobTracker提交作业这个最简单的客户端服务器交互的代码中，去跟踪和了解下RPC是怎么被使用的。不同于准备发表博客时搜索的几篇博文，试图通过一种具体的场景来介绍，属于比较初级。其他DataNode和Namenode之间，Tasktracker和JobTracker之间的交互基本也都一样。为了引用的代码篇幅尽可能少，忽略了代码中写日志（包括Metrics）、某些判断等辅助代码。

1 RPC客户端请求（从JobClient 的jobSubmitClient 入手）
　　Jobclient包含一个JobSubmissionProtocol  jobSubmitClient类型的句柄，从作业提交一节的介绍中看到Jobclient的计划所有重要操作都是通过jobSubmitClient来完成的。包括







　　所有这些方法都在JobSubmissionProtocol接口中定义。在0.20.1的时候已经到Version 20了，在2.2.0好像到了Version 40了,说明功能一直在增强。






　　客户端的某个方法调用如何会调用到服务端的方法呢？在客户端机器上调用JobClient的getAllJobs(),怎么调用到了服务端JobTracker的getAllJobs()。这也是我尝试讲明白的核心内容。为了体现代码的一步一步分析总结在最后。可能循序渐进的作用没起到，还会笔记读起来笔记乱，感受有点不太好可能:-(。
　　首先看客户端JobClient中的jobSubmitClient初始化方法。在JobClient的init方法中判断不是local的方式则会调用createRPCProxy方法，进而调用RPC的getProxy方法。方法连接对应IP的服务器。比较客户端和服务端的RPC版本一致，返回一个JobSubmissionProtocol类型的句柄，抛出VersionMismatch异常。





private JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
Configuration conf) throws IOException {
return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
JobSubmissionProtocol.versionID, addr, getUGI(conf), conf,
NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
}
public static VersionedProtocol getProxy(Class protocol,
long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
Configuration conf, SocketFactory factory) throws IOException {   
VersionedProtocol proxy =
(VersionedProtocol) Proxy.newProxyInstance(
protocol.getClassLoader(), new Class[] { protocol },
new Invoker(addr, ticket, conf, factory));
long serverVersion = proxy.getProtocolVersion(protocol.getName(),
clientVersion);
if (serverVersion == clientVersion) {
return proxy;
} else {
throw new VersionMismatch(protocol.getName(), clientVersion,
serverVersion);
}
}
getProxy 　　注意到调用了java的反射代理，在构建VersionedProtocol的时候Proxy.newProxyInstance方法初始化了一个Invoker类型的对象。
　　该对象是org.apache.hadoop.ipc.RPC.包下Server类的一个内部类。



static class Invoker implements InvocationHandler
　　这下明白了！基于java的reflect机制提供的一种Proxy使用方式。InvocationHandler这个Interface的作用就是把proxy 上的方法调用派发到实现了InvocationHandler的类上来。即Jobclient上中jobSubmitClient的任何调用都会派发到这个Invoker上来。
　　那么Invoker中做了什么事情呢？Invoker类实现了InvocationHandler接口定义的唯一的invoke方法。只是把传入的调用信息，包括方面名，方法参数封装为一个invocation对象，调用用Client  client对象的call方法来执行操作。





public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
ObjectWritable value = (ObjectWritable)
client.call(new Invocation(method, args), address,
method.getDeclaringClass(), ticket);
return value.get();}
iInvoker nvoke　　了解Client的call方法，该方法的主要作用是把参数发送给指定服务端地址上的IPC server。并获取结果。构建一个Call对象，封装了请求参数（其实是Invocation封装了方法和参数的对象），创建一个连接到IPC服务器的connection，然后发送出去。（client发送请求还是有些业务的，包括Client下的几个内部类的工作，在此略去）





    public Writable call(Writable param, InetSocketAddress addr,
Class protocol, UserGroupInformation ticket)
{
Call call = new Call(param);
Connection connection = getConnection(addr, protocol, ticket, call);
connection.sendParam(call); // send the parameter
return call.value;
}
Client call()　　客户端的主要过程总结如下：客户端Jobclient的创建一个JobSubmissionProtocol
　　jobSubmitClient，jobSubmitClient的所有请求都会通过Invoker封装成一个请求，通过Client的call方法发送到服务端。

2 RPC服务端处理(看Jobtracker的interTrackerServer响应请求)
　　接下来看法服务器是如何接收请求，Client的call将请求发送到什么样的服务器？服务器如何解释这些请求，如何响应请求的。
　　服务端JobTracker实现了JobSubmissionProtocol接口，因此提供了JobSubmissionProtocol定义的所有方法



public class JobTracker implements MRConstants, InterTrackerProtocol,
JobSubmissionProtocol, TaskTrackerManager, RefreshAuthorizationPolicyProtocol
　　在JobTracker内包含一个类型org.apache.hadoop.ipc.Server的的实例interTrackerServer ，该实例其实是响应客户端的的RPC调用的服务实例。



this.interTrackerServer = RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount, false, conf);
　　查看RPC的getServer方法





public static Server getServer(final Object instance, final String bindAddress, final int port,
final int numHandlers,
final boolean verbose, Configuration conf)
throws IOException {
return new Server(instance, conf, bindAddress, port, numHandlers, verbose);
}
RPC getServer　　再往下看其实Server的构造函数，就是在某个Ip和端口上监听，响应客户端发起的请求。多么典型的客户端服务器模式呀。代码看上去多么想Socket通信那一套呀。看到了bindAddress，看到了port，还看到socketSendBufferSize。没错！





protected Server(String bindAddress, int port,
Class