【hadoop代码笔记】通过JobClient对Jobtracker的调用详细了解Hadoop RPC
Hadoop的各个服务间，客户端和服务间的交互采用RPC方式。关于这种机制介绍的资源很多，也不难理解，这里不做背景介绍。只是尝试从Jobclient向JobTracker提交作业这个最简单的客户端服务器交互的代码中，去跟踪和了解下RPC是怎么被使用的。不同于准备发表博客时搜索的几篇博文，试图通过一种具体的场景来介绍，属于比较初级。其他DataNode和Namenode之间，Tasktracker和JobTracker之间的交互基本也都一样。为了引用的代码篇幅尽可能少，忽略了代码中写日志（包括Metrics）、某些判断等辅助代码。
1 RPC客户端请求（从JobClient的jobSubmitClient 入手）
Jobclient包含一个JobSubmissionProtocol jobSubmitClient类型的句柄，从作业提交一节的介绍中看到Jobclient的计划所有重要操作都是通过jobSubmitClient来完成的。包括
JobSubmissionProtocol Outline
所有这些方法都在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异常。
1private JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
2 Configuration conf) throws IOException {
3 return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
4 JobSubmissionProtocol.versionID, addr, getUGI(conf), conf,
5 NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
6 }
7 public static VersionedProtocol getProxy(Class< > protocol,
8 long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
9 Configuration conf, SocketFactory factory) throws IOException {
10
11 VersionedProtocol proxy =
12 (VersionedProtocol) Proxy.newProxyInstance(
13 protocol.getClassLoader(), new Class[] { protocol },
14 new Invoker(addr, ticket, conf, factory));
15 long serverVersion = proxy.getProtocolVersion(protocol.getName(),
16 clientVersion);
17 if (serverVersion == clientVersion) {
18 return proxy;
19 } else {
20 throw new VersionMismatch(protocol.getName(), clientVersion,
21 serverVersion);
22 }
23}
注意到调用了java的反射代理在构建VersionedProtocol的时候ProxynewProxyInstance方法初始化了一个Invoker类型的对象。 该对象是org.apache.hadoop.ipc.RPC.包下Server类的一个内部类。
1static class Invoker implements InvocationHandler
这下明白了！基于java的reflect机制提供的一种Proxy使用方式。InvocationHandler这个Interface的作用就是把proxy 上的方法调用派发到实现了InvocationHandler的类上来。即Jobclient上中jobSubmitClient的任何调用都会派发到这个Invoker上来。
那么Invoker中做了什么事情呢？Invoker类实现了InvocationHandler接口定义的唯一的invoke方法。只是把传入的调用信息，包括方面名，方法参数封装为一个invocation对象，调用用Client client对象的call方法来执行操作。
1public Object invoke(Object proxy, Method method, Object[] args)
2 throws Throwable {
3 ObjectWritable value = (ObjectWritable)
4 client.call(new Invocation(method, args), address,
5 method.getDeclaringClass(), ticket);
6 return value.get();}
了解Client的call方法，该方法的主要作用是把参数发送给指定服务端地址上的IPC server。并获取结果。构建一个Call对象，封装了请求参数（其实是Invocation封装了方法和参数的对象），创建一个连接到IPC服务器的connection，然后发送出去。（client发送请求还是有些业务的，包括Client下的几个内部类的工作，在此略去）
1public Writable call(Writable param, InetSocketAddress addr,
2 Class< > protocol, UserGroupInformation ticket)
3 {
4 Call call = new Call(param);
5 Connection connection = getConnection(addr, protocol, ticket, call);
6 connection.sendParam(call); // send the parameter
7 return call.value;
8 }
客户端的主要过程总结如下：客户端Jobclient的创建一个JobSubmissionProtocol jobSubmitClient，jobSubmitClient的所有请求都会通过Invoker封装成一个请求，通过Client的call方法发送到服务端。
2 RPC服务端处理(看Jobtracker的interTrackerServer响应请求)
接下来看法服务器是如何接收请求，Client的call将请求发送到什么样的服务器？服务器如何解释这些请求，如何响应请求的。
服务端JobTracker实现了JobSubmissionProtocol接口，因此提供了JobSubmissionProtocol定义的所有方法
1public class JobTracker implements MRConstants, InterTrackerProtocol,
2 JobSubmissionProtocol, TaskTrackerManager, RefreshAuthorizationPolicyProtocol
在JobTracker内包含一个类型org.apache.hadoop.ipc.Server的的实例interTrackerServer ，该实例其实是响应客户端的的RPC调用的服务实例。
1this.interTrackerServer = RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount, false, conf);
查看RPC的getServer方法
1public static Server getServer(final Object instance, final String bindAddress, final int port,
2 final int numHandlers,
3 final boolean verbose, Configuration conf)
4 throws IOException {
5 return new Server(instance, conf, bindAddress, port, numHandlers, verbose);
6 }
再往下看其实Server的构造函数，就是在某个Ip和端口上监听，响应客户端发起的请求。多么典型的客户端服务器模式呀。代码看上去多么想Socket通信那一套呀。看到了bindAddress，看到了port，还看到socketSendBufferSize。没错！
1protected Server(String bindAddress, int port,
2 Class< extends Writable> paramClass, int handlerCount,
3 Configuration conf, String serverName)
4 throws IOException {
5 this.bindAddress = bindAddress;
6 this.conf = conf;
7 this.port = port;
8 this.paramClass = paramClass;
9 this.handlerCount = handlerCount;
10 this.socketSendBufferSize = 0;
11 this.maxQueueSize = handlerCount * MAX_QUEUE_SIZE_PER_HANDLER;
12 this.callQueue = new LinkedBlockingQueue<Call>(maxQueueSize);
13 this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
14 this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
15 this.thresholdIdleConnections = conf.getInt("ipc.client.idlethreshold", 4000);
16
17 // Start the listener here and let it bind to the port
18 listener = new Listener();
19 this.port = listener.getAddress().getPort();
20 this.rpcMetrics = new RpcMetrics(serverName,
21 Integer.toString(this.port), this);
22 this.tcpNoDelay = conf.getBoolean("ipc.server.tcpnodelay", false);
23
24 // Create the responder here
25 responder = new Responder();
26 }
同时不小心注意到Server类的outline阵容还是很宏大的，除了一长串的方法外，还包括Call, Connection, Handler,Listener, responder 五个内部类，猜就是这些协作来完成Server的服务响应处理。
1private BlockingQueue<Call> callQueue; // queued calls
2 private List<Connection> connectionList =
3 Collections.synchronizedList(new LinkedList<Connection>());
4 private Listener listener = null;
5 private Responder responder = null;
6 private Handler[] handlers = null;
再看看Server的start（）方法
1public synchronized void start() throws IOException {
2 responder.start();
3 listener.start();
4 handlers = new Handler[handlerCount];
5
6 for (int i = 0; i < handlerCount; i++) {
7 handlers[i] = new Handler(i);
8 handlers[i].start();
9 }
10}
其中，在Server的构造函数中看到了两个差不多能猜到其功能的东西：Listener & Responder。从命名上几乎就能猜到，他们分别是监听用户请求和响应用户请求的线程？应该是线程吧？居然猜对了！
先看下Listener。构造函数如下
1public Listener() throws IOException {
2 address = new InetSocketAddress(bindAddress, port);
3 // Create a new server socket and set to non blocking mode
4 acceptChannel = ServerSocketChannel.open();
5 acceptChannel.configureBlocking(false);
6
7 // Bind the server socket to the local host and port
8 bind(acceptChannel.socket(), address, backlogLength);
9 port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
10 // create a selector;
11 selector= Selector.open();
12
13 // Register accepts on the server socket with the selector.
14 acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
15}
重点看下线程的业务方法，即其run方法做了些啥。方法虽然很长，但是业务很典型，在服务端监听，收到数据就接收。
1public void run() {
2 SERVER.set(Server.this);
3 while (running) {
4 SelectionKey key = null;
5 selector.select();
6 Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
7 while (iter.hasNext()) {
8 key = iter.next();
9 iter.remove();
10 try {
11 if (key.isValid()) {
12 if (key.isAcceptable())
13 doAccept(key);
14 else if (key.isReadable())
15 doRead(key);
16 }
17
18 }
接着看下接受数据的 doAccept(SelectionKey key)和doRead(SelectionKey key)方法。
doAccept做的事情是把每一个数据连接的请求绑定到一个Connection对象上，并把Connection全部添加到connectionList集合中；doRead做的事情是对每个Connection执行readAndProcess操作。
1void doAccept(SelectionKey key)
2 {
3 Connection c = null;
4 ServerSocketChannel server = (ServerSocketChannel) key.channel();
5 // accept up to 10 connections
6 for (int i=0; i<10; i++) {
7 SocketChannel channel = server.accept();
8 SelectionKey readKey = channel.register(selector, SelectionKey.OP_READ);
9 c = new Connection(readKey, channel, System.currentTimeMillis());
10 readKey.attach(c);
11 synchronized (connectionList) {
12 connectionList.add(numConnections, c);
13 numConnections++;
14 }
15 }
16
17 void doRead(SelectionKey key)
18 {
19 Connection c = (Connection)key.attachment();
20 c.setLastContact(System.currentTimeMillis());
21 count = c.readAndProcess();
22 }
需要关注下org.apache.hadoop.ipc.Server.Connection类。重点看listener doRead 中调用的readAndProcess方法
1data = ByteBuffer.allocate(dataLength);
2 count = channelRead(channel, data);
3
4 if (headerRead) {
5 processData();
6 data = null;
7 return count;
8 } else {
9 processHeader();
10 headerRead = true;
11 data = null;
12 }
13 authorize(user, header);
其中的processHeader()作用是解析出通信的protocol类
1header.readFields(in);
2String protocolClassName = header.getProtocol();
3protocol = getProtocolClass(header.getProtocol(), conf);
processData的主要代码如下：
1int id = dis.readInt();
2Writable param = ReflectionUtils.newInstance(paramClass, conf); param.readFields(dis);
3 Call call = new Call(id, param, this);
4callQueue.put(call);
读取调用Id，从读取的数据中构建参数，并构造一个Call对象，放置到BlockingQueue
1private boolean processResponse(LinkedList<Call> responseQueue,
2 boolean inHandler)
3{
4call = responseQueue.removeFirst();
5 SocketChannel channel = call.connection.channel;
6 int numBytes = channelWrite(channel, call.response);
7}
Handler是处理请求的线程。是对队列中的每个call进行处理的类。前面看Server包含的实例的时候看到了，Server包含一个Handler数组，在Server的start方法中启动了Listener，Responder线程，同时初始化了handlerCount个Handler线程并且启动。 主要还是看run方法。主要是从请求队列callQueue中逐个取出call了，并进行处理。处理过程即，对每个call，执行Server的call[ 方法，（实际的call方法是从org.apache.hadoop.ipc.RPC.Server，继承了org.apache.hadoop.ipc.Server，不是一个Server哦！这个在后面RPC类中会讲到）并调用Responder方法doRespond，把结果返回。
1while (running)
2{
3 final Call call = callQueue.take(); // pop the queue; maybe blocked here
4 CurCall.set(call);
5 value = Subject.doAs(call.connection.user,
6 new PrivilegedExceptionAction<Writable>() {
7 @Override
8 public Writable run() throws Exception {
9 // make the call
10 return call(call.connection.protocol,
11 call.param, call.timestamp);
12 }
13 }
14 );
15 CurCall.set(null);
16 setupResponse(buf, call,
17 (error == null) Status.SUCCESS : Status.ERROR,
18 value, errorClass, error);
19 responder.doRespond(call);
20 }
调用的setupResponse方法
1private void setupResponse(ByteArrayOutputStream response,
2 Call call, Status status,
3 Writable rv, String errorClass, String error)
4
5DataOutputStream out = new DataOutputStream(response);
6 out.writeInt(call.id); // write call id
7 out.writeInt(status.state); // write status
8 rv.write(out);
9call.setResponse(ByteBuffer.wrap(response.toByteArray()));
核心就一句，把执行结果写到Call中去。 顺便看下上面方法调用的Responder的doRespond方法，即把经过handler处理的带有结果的call放到对应的响应队列中，等待responder线程来逐个返回给客户端，注意看到一个，如果队列中只有一个对象时，直接调用processResponse触发把结果翻过给客户端。
1void doRespond(Call call) throws IOException {
2 {
3 call.connection.responseQueue.addLast(call);
4 if (call.connection.responseQueue.size() == 1) {
5 processResponse(call.connection.responseQueue, true);
6 }
7 }
8}
即handler完成call之后就开始向客户端写call结果，但是结果可能太多，无法通过一次性发送完毕，而发送之后还要等待client接受完毕才能再发，如果现在handler在那里等待客户端接受完毕，然后再发，效率不高。解决办法是handler处理完毕之后，只向client发送一次处理结果。如果这一次将处理结果发送完毕，接下来就没有response的事情了，如果没有发送完毕，接下来response负责将剩下的处理结果发送给客户端。这样handler的并发量会大一些。详细可参照Responder线程的run方法和
1writeSelector.select(PURGE_INTERVAL);
2 Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
3 while (iter.hasNext()) {
4 SelectionKey key = iter.next();
5 iter.remove();
6 if (key.isValid() && key.isWritable()) {
7 doAsyncWrite(key);
8 }
//在doAsyncWrite方法中，从key中获得Call，并对每个call执行processResponse方法。
1private void doAsyncWrite(SelectionKey key)
2 Call call = (Call)key.attachment();
3 processResponse(call.connection.responseQueue, false))
至此观察到服务端的工作的主要过程是：
-
Server启动的时候，启动一个listener线程，一个Responder线程，若干个Handler线程。
-
Listener线程接受客户端发起的请求（在doAccept中接收请求，并且每个请求构建一个Connection，绑定到一个SelectionKey上），读取请求数据，根据请求数据构造call对象，将Call加入队列。
-
Handler线程从请求队列（callQueue）中获取每个Call，进行处理，把处理结果放到对应的connection的应答队列中）（responseQueue，通过调用responder.doRespond）。Responder线程检查负责把结果返回给客户端。（processResponse，把responseQueue队列的结果数据返回）
有一点需要继续关注一下，就是Handler中处理了客户端发起的请求，并且将结果通过Responder返回。但是并没有发现Handler是调用到了Jobtracker的方法。需要继续向下多看一点即可。 从代码看Handler的call方法调用的是org.apache.hadoop.ipc.Server.的抽象方法
1public abstract Writable call(Class< > protocol, Writable param, long receiveTime)
实际调用是org.apache.hadoop.ipc.Server的子类org.apache.hadoop.ipc.RPC.Server的call方法从org.apache.hadoop.ipc.RPC.Server的call方法入手，该类在是RPC类的一个静态内部类。
1//传入的param其实是一个Invocation对象。根据该对象的方面明，参数声明构造Method，调用Method，得到执行结果，根据返回值得类型，构造一个Writable的对象。
2Writable call(Class< > protocol, Writable param, long receivedTime)
3{
4Invocation call = (Invocation)param;
5 Method method =
6 protocol.getMethod(call.getMethodName(),
7 call.getParameterClasses());
8 method.setAccessible(true);
9 Object value = method.invoke(instance, call.getParameters());
10 return new ObjectWritable(method.getReturnType(), value);
11}
重点看这一句
1 Object value = method.invoke(instance, call.getParameters());
即最终是调用该instance上的对应名称的方法。而instance是那个实例呢？而从Server的构造方法中得到答案。
1this.interTrackerServer = RPC.getServer(this, addr.getHostName(), addr.getPort(), handlerCount, false, conf);</pre>
即最终调用到JobTracker的对应方法。
3 主要流程总结
整个调用过程总结如下：根据接口JobSubmissionProtocol动态代理生成一个代理对象jobSubmitClient，调用这个代理对象的时候;用户的调用请求被RPC的Invoker捕捉到，然后包装成调用请求，序列化成数据流发送到服务端Jobtracker的interTrackerServer实例；服务端interTrackerServer从数据流中解析出调用请求，然后根据用户所希望调用的接口JobSubmissionProtocol，通过反射调用接口真正的实现对象Jobtracker，再把调用结果返回给客户端的jobSubmitClient。
4 主要类功能描述
至此根据Jobclient通过RPC方式向JobTracker请求服务的过程就描述完毕，到此主要内容应该介绍完毕。但是看到cover的代码，发现RPC的主要功能在里面了。
为了功能完整期间，在动态的串联这些类以为，把涉及到主要类的功能做个描述，其实大部分在前面代码中也有提到。
RPC类是对Server、Client的具体化。在RPC类中规定，客户程序发出请求调用时，参数类型必须是Invocation；从服务器返回的值类型必须是ObjectWritable。RPC类是对Server、Client的包装，简化用户的使用。如果一个类需充当服务器，只需通过RPC类的静态方法getServer获得Server实例，然后start。同时此类提供协议接口的实现。如果一个类充当客户端，可以通过getProxy或者waitForProxy获得一个实现了协议接口的proxy object，与服务器端交互。
- Server.Listener： RPC Server的监听者，用来接收RPC Client的连接请求和数据，其中数据封装成Call后PUSH到Call队列。
- Server.Handler： RPC Server的Call处理者，和Server.Listener通过Call队列交互。
- Server.Responder： RPC Server的响应者。Server.Handler按照异步非阻塞的方式向RPC Client发送响应，如果有未发送出的数据，交由Server.Responder来完成。
- Server.Connection： RPC Server数据接收者。提供接收数据，解析数据包的功能。
- Server.Call： 持有客户端的Call信息。 org.apache.hadoop.ipc.Client
- Client.ConnectionId：到RPC Server对象连接的标识
- Client.Call： Call调用信息。
- Client.ParallelResults： Call响应。 org.apache.hadoop.ipc.RPC
- RPC.Invoker 对InvocationHandler的实现，提供invoke方法，实现RPC Client对RPC Server对象的调用。 RPC.Invocation 用来序列化和反序列化RPC Client的调用信息。（主要应用JAVA的反射机制和InputStream/OutputStream）
5 VersionedProtocol的其他子接口
除了JobClient和Jobtracker之间通信的JobSubmissionProtocol外，最后查看下VersionedProtocol 的继承树
Hadoop中主要服务进程分别实现了各种接口，进而向外提供各种服务，其客户端通过RPC调用对应的服务。当然此处的客户端只是指调用上的客户端。 VersionedProtocol ：它是所有RPC协议接口的父接口，只有一个方法：getProtocolVersion（）。其子类接口的功能分别如下。 HDFS相关：
- ClientDatanodeProtocol ：一个客户端和datanode之间的协议接口，用于数据块恢复
- ClientProtocol ：client与Namenode交互的接口，所有控制流的请求均在这里，如：创建文件、删除文件等；
- DatanodeProtocol : Datanode与Namenode交互的接口，如心跳、blockreport等；
- NamenodeProtocol ：SecondaryNode与Namenode交互的接口。 Mapreduce相关
- InterDatanodeProtocol ：Datanode内部交互的接口，用来更新block的元数据；
- InnerTrackerProtocol ：TaskTracker与JobTracker交互的接口，功能与DatanodeProtocol相似；
- JobSubmissionProtocol ：JobClient与JobTracker交互的接口，用来提交Job、获得Job等与Job相关的操作；
- TaskUmbilicalProtocol ：Task中子进程与母进程交互的接口，子进程即map、reduce等操作，母进程即TaskTracker，该接口可以回报子进程的运行状态。