Hadoop mr heartbeat rpc代码分析
一、概述
Hadoop中mr类图大致如下所示，其中只是简单列出了一些主要的功能模块。
JobConf、JobTracker、TaskTracker、RPC Server等组件。在图中，我用红色框框框出了一些基本的类。此图基本反映了MR的类图结构。
一个简单的job在hadoop上面跑起来，基本可以分为10个步骤。如下图所示：
其中第7条线heartbeat应该是比较复杂且具有代表性的。接下来，我主要分析下，mr心跳的具体流程。（包括RPC）
二、分析heartbeat及RPC
二、分析heartbeat及RPC
hadoop是master-slave结构，所以需要slave报告自身的情况。为了得到一个HeartbeatResponse，大致需要经过以下的流程。
上图是一个序列图,是代码实现的调用序列，我们简化为流程图如下：
我们大致认为一次心跳（包括RPC）需要经过7个步骤。在这个7个步骤中，有大约6个主要的线程在服务。
TaskTracker端的：TaskTreacker,Client.Connection
JobTracker端的：Server.Listener,Server.Reader,Server.Handler,Server.Responder
7个步骤为：
1)、TaskTracker定期发送请求。TaskTracker持有一个代理对象InterTrackerProtocol.此对象经过动态代理处理,实际上是调用RPC.Invoker类。此类再调用 client.call()。此方法中，会实例化一个Client.Connection对象，并启动此对象。发送远程请求后，TaskTracker会阻塞在Call上面，等待Client.Connection线程被唤醒后调用call.notify()。
Call call = new Call(param);
Connection connection = getConnection(addr, ticket, call);
connection.sendParam(call); // send the parameter
synchronized (call) {
while (!call. done) {
try {
call.wait(); // wait for the result
} catch (InterruptedException ignored) {}
}
注意Client.Connection线程一直阻塞在DataInputStream.readInt()等待JobTracker的响应。
图：Client.Connection线程
2)、Server.Listener会监听到请求并建立连接，再调用Reader来做读取网络数据。与Server.Reader的协同主要是通过adding变量、Server.Reader.wait()、readSelector;完成后阻塞在selector.select();
while ( adding) {
this.wait(1000);
}
图:Server.Listener线程
3)、Server.Reader主要是读取数据，做一些分析，再生成Server.call放进BlockingQueue<Call> callQueue队列中。完成后会阻塞在readSelector.select();
图:Server.Reader线程
4)、Server.Handler阻塞在队列上面，当callQueue队列中存在数据的时候就会被唤醒。执行任务主要是 调用 Server的子类（RPC.Server）���现的Call()函数。此函数会根据TaskTreacker传入的函数来反射调用JobTracker的实现（这里是InterTrackerProtocol接口的实现heartbeat(x,x,x)），完成后，再调用Server.Responder的函数。在一些特别的条件下会与Server.Responder线程协调工作。由Server.Responder线程完成数据压入网络的工作。
图:Server.Handler线程
5)、Server.Handler线程会反射调用TaskTreacker.heartbeat(x,x,x)会执行相关的工作，拿到结果HeartbeatResponse。
6)、Server.Responder线程在一些情况下会把响应结果压入网络中。
7)、Client.Connection线程被唤醒，他再调用call.nodify()唤醒TaskTreacker主线程。
此点，在开始Connection发送请求的时候，会把call的id存入在map中，在被唤醒的时候DataInputStream.readInt()直接返回一个call的id。用id到map查找相应的call，再调用call.nodify()唤醒TaskTracker,TaskTracker被唤醒后再根据call.value做相应的事情。如：启动一个任务。
备注：
其实在心跳中，最复杂的一部分还是RPC的实现,分为大约6组线程的协调。以上的代码分析是基于0.19.1，高版本的代码可能所有不同。
三、版权信息
作者工作于阿里巴巴 封神
相关推荐
graseed 2020-10-28
SXIAOYI 2020-09-16
jinhao 2020-09-07
yoohsummer 2020-06-01
impress 2020-02-20
chenshuixian 2013-06-01
羽化大刀Chrome 2013-05-31
kanpiaoxue 2013-06-19
quanhaoH 2013-06-17
pigsmall 2020-11-19
大数据杂谈 2020-09-26
ChinaWin 2020-08-13
mohanzb 2020-08-01
王国平 2020-06-20
kangtingting0 2020-05-20
MichelinMessi 2020-02-19
nicepainkiller 2020-01-25
hfszy0 2013-05-15
lizhenmxcz 2013-05-12