共计 16110 个字符，预计需要花费 41 分钟才能阅读完成。
在 Hadoop 中 JT（JobTracker）与 TT（TaskTracker）之间的通信是通过心跳机制完成的。JT 实现 InterTrackerProtocol 协议，该协议定义了 JT 与 TT 之间的通信机制——心跳。心跳机制实际上就是一个 RPC 请求，JT 作为 Server，而 TT 作为 Client，TT 通过 RPC 调用 JT 的 heartbeat 方法，将 TT 自身的一些状态信息发送给 JT，同时 JT 通过返回值返回对 TT 的指令。
目录 ：
Hadoop1.2.1 源码解析系列:JT 与 TT 之间的心跳通信机制——TT 篇 http://www.linuxidc.com/Linux/2014-06/103216.htm
Hadoop1.2.1 源码解析系列：JT 与 TT 之间的心跳通信机制——JT 篇 http://www.linuxidc.com/Linux/2014-06/103217.htm
Hadoop1.2.1 源码解析系列：JT 与 TT 之间的心跳通信机制——命令篇 http://www.linuxidc.com/Linux/2014-06/103218.htm
心跳有三个作用：
1）判断 TT 是否活着
2）报告 TT 的资源情况以及任务运行情况
3）为 TT 发送指令（如运行 task，kill task 等）
下面详细阅读下涉及到心跳调用的源码。
首先我们需要清楚，心跳机制是 TT 调用 JT 的方法，而非 JT 主动调用 TT 的方法。TT 通过 transmitHeartBeat 方法调用 JT 的 heartbeat 方法。
1.TaskTracker.transmitHeartBeat：
// Send Counters in the status once every COUNTER_UPDATE_INTERVAL
boolean sendCounters;
if (now > (previousUpdate + COUNTER_UPDATE_INTERVAL)) {
sendCounters = true;
previousUpdate = now;
}
else {
sendCounters = false;
}
根据 sendCounters 的间隔判断此次心跳是否发送计算器信息。
2.TaskTracker.transmitHeartBeat：
1.TaskTracker.transmitHeartBeat：
// Check if the last heartbeat got through…
// if so then build the heartbeat information for the JobTracker;
// else resend the previous status information.
//
if (status == null) {
synchronized (this) {
status = new TaskTrackerStatus(taskTrackerName, localHostname,
httpPort,
cloneAndResetRunningTaskStatuses(
sendCounters),
taskFailures,
localStorage.numFailures(),
maxMapSlots,
maxReduceSlots);
}
} else {
LOG.info(“Resending ‘status’ to ‘” + jobTrackAddr.getHostName() +
“‘ with reponseId ‘” + heartbeatResponseId);
}
此处根据 status 变量是否为 null，判断上次的心跳是否成功发送。tatus!=null，则表示上次的心跳尚未发送，所以直接将上次收集到的 TT 状态信息（封装在 status 中）发送给 JT；相反，status==null，则表示上次心跳已完成，重新收集 TT 的状态信息，同样封装到 status 中。下面详细看下 new TaskTrackerStatus（）方法。注意此处有个 cloneAndResetRunningTaskStatuses(sendCounters) 方法：
private synchronized List<TaskStatus> cloneAndResetRunningTaskStatuses(
boolean sendCounters) {
List<TaskStatus> result = new ArrayList<TaskStatus>(runningTasks.size());
for(TaskInProgress tip: runningTasks.values()) {
TaskStatus status = tip.getStatus();
status.setIncludeCounters(sendCounters);
// send counters for finished or failed tasks and commit pending tasks
if (status.getRunState() != TaskStatus.State.RUNNING) {
status.setIncludeCounters(true);
}
result.add((TaskStatus)status.clone());
status.clearStatus();
}
return result;
}
该方法中涉及到 runningTasks 队列，该队列保存了该 TT 上接收的所有未完成的 Task 任务，通过 runningTasks.values() 可以获取 TT 当前所有未完成的 Task，然后获取每个 TaskInProgress 的 status 信息，同时根据第一步判断出的 sendCounters（true/false）决定是否发送 counters 信息（includeCounters），即是否将 counters 对象序列化到 TaskStatus 对象中，这里需要注意如果 TaskInProgress 不处于 Running 状态，则 includeCounters 设为 true，即发送 counters 信息。
————————————– 分割线 ————————————–
Ubuntu 13.04 上搭建 Hadoop 环境 http://www.linuxidc.com/Linux/2013-06/86106.htm
Ubuntu 12.10 +Hadoop 1.2.1 版本集群配置 http://www.linuxidc.com/Linux/2013-09/90600.htm
Ubuntu 上搭建 Hadoop 环境（单机模式 + 伪分布模式）http://www.linuxidc.com/Linux/2013-01/77681.htm
Ubuntu 下 Hadoop 环境的配置 http://www.linuxidc.com/Linux/2012-11/74539.htm
单机版搭建 Hadoop 环境图文教程详解 http://www.linuxidc.com/Linux/2012-02/53927.htm
Hadoop LZO 安装教程 http://www.linuxidc.com/Linux/2013-01/78397.htm
Hadoop 集群上使用 Lzo 压缩 http://www.linuxidc.com/Linux/2012-05/60554.htm
————————————– 分割线 ————————————–
3.TaskTrackerStatus()：
public TaskTrackerStatus(String trackerName, String host,
int httpPort, List<TaskStatus> taskReports,
int taskFailures, int dirFailures,
int maxMapTasks, int maxReduceTasks) {
this.trackerName = trackerName;
this.host = host;
this.httpPort = httpPort;
this.taskReports = new ArrayList<TaskStatus>(taskReports);
this.taskFailures = taskFailures;
this.dirFailures = dirFailures;
this.maxMapTasks = maxMapTasks;
this.maxReduceTasks = maxReduceTasks;
this.resStatus = new ResourceStatus();
this.healthStatus = new TaskTrackerHealthStatus();
}
这里只是进行简单的变量复制操作，分析下其中一些参数的含义：
1）taskReports：包含该 TT 上目前所有的 Task 状态信息，其中的 counters 信息会根据之前判断 sendCounters 值进行决定是否发送，上一步有提到。
2）taskFailures：该 TT 上失败的 Task 总数（重启会清空），该参数帮助 JT 决定是否向该 TT 提交 Task，因为失败数越多表明该 TT 可能出现 Task 失败的概率越大。
3）dirFailures：这个值是 mapred.local.dir 参数设置的目录中有多少是不可用的（以后会详细提到）
4）maxMapSlots/maxReduceSlots：这个值是 TT 可使用的最大 map 和 reduce slot 数量
初始化完成，继续回到 TaskTracker.transmitHeartBeat 方法。
4.TaskTracker.transmitHeartBeat：
// Check if we should ask for a new Task
//
boolean askForNewTask;
long localMinSpaceStart;
synchronized (this) {
askForNewTask =
((status.countOccupiedMapSlots() < maxMapSlots ||
status.countOccupiedReduceSlots() < maxReduceSlots) &&
acceptNewTasks);
localMinSpaceStart = minSpaceStart;
}
if (askForNewTask) {
askForNewTask = enoughFreeSpace(localMinSpaceStart);
long freeDiskSpace = getFreeSpace();
long totVmem = getTotalVirtualMemoryOnTT();
long totPmem = getTotalPhysicalMemoryOnTT();
long availableVmem = getAvailableVirtualMemoryOnTT();
long availablePmem = getAvailablePhysicalMemoryOnTT();
long cumuCpuTime = getCumulativeCpuTimeOnTT();
long cpuFreq = getCpuFrequencyOnTT();
int numCpu = getNumProcessorsOnTT();
float cpuUsage = getCpuUsageOnTT();
status.getResourceStatus().setAvailableSpace(freeDiskSpace);
status.getResourceStatus().setTotalVirtualMemory(totVmem);
status.getResourceStatus().setTotalPhysicalMemory(totPmem);
status.getResourceStatus().setMapSlotMemorySizeOnTT(
mapSlotMemorySizeOnTT);
status.getResourceStatus().setReduceSlotMemorySizeOnTT(
reduceSlotSizeMemoryOnTT);
status.getResourceStatus().setAvailableVirtualMemory(availableVmem);
status.getResourceStatus().setAvailablePhysicalMemory(availablePmem);
status.getResourceStatus().setCumulativeCpuTime(cumuCpuTime);
status.getResourceStatus().setCpuFrequency(cpuFreq);
status.getResourceStatus().setNumProcessors(numCpu);
status.getResourceStatus().setCpuUsage(cpuUsage);
}
从源码中的注释可以知道，此处是 TT 根据自身资源使用情况判断是否接收 new task。
更多详情见请继续阅读下一页的精彩内容 ：http://www.linuxidc.com/Linux/2014-06/103216p2.htm
首先第一步 status.countOccupiedMapSlots() 获得该 TT 上已占用的 map slot 数量：
/**
* Get the number of occupied map slots.
* @return the number of occupied map slots
*/
public int countOccupiedMapSlots() {
int mapSlotsCount = 0;
for (TaskStatus ts : taskReports) {
if (ts.getIsMap() && isTaskRunning(ts)) {
mapSlotsCount += ts.getNumSlots();
}
}
return mapSlotsCount;
}
方法内部是根据 taskReports 中的 TaskStatus 进行判断，这里计算的是 map slot，所以会判断 ts.getIsMap()，如果该 task 是 map 任务，且 isTaskRunning() 返回 true，则获取该 task 所需的 slot 数量。isTaskRunning() 方法内部判断逻辑是：该 task 处于 RUNNING 或者 UNASSIGNED 状态，或者处于 CleanerUp 阶段（这里可能是 Task 处于 FAILED_UNCLEAN 或者 KILLED_UNCLEAN 阶段）。这个方法会计算出 TT 当前已占用的 map slot 数量。同样的通过 countOccupiedReduceSlots() 方法计算出 TT 当前已占用的 reduce slot 数量。获取到 occupied map/reduce slots 后将其同 maxMapSlots/maxReduceSlots 进行比较，这里是“||”而非“&&”，表示只要有 map slot 或者有 reduce slot 就可以接收新任务，当然还需要满足 acceptNewTasks==true 的条件。acceptNewTasks 会在其他地方根据 TT 可使用的空间进行合适的赋值。以上可以判断出是否可以接收新任务，即 askForNewTask 值。
localMinSpaceStart = minSpaceStart，minSpaceStart 由 mapred.local.dir.minspacestart 参数决定，默认是 0，即无限制，该值的意思应该是可接收新任务的 localDirs 最小的可用空间大小。接下来可以看到该值能够影响 acceptNewTasks 值。
当 acceptNewTasks==true 时，即初步判断可以接收新任务，会再次根据 localMinSpaceStart 判断是否可接收新任务。
/**
* Check if any of the local directories has enough
* free space (more than minSpace)
*
* If not, do not try to get a new task assigned
* @return
* @throws IOException
*/
private boolean enoughFreeSpace(long minSpace) throws IOException {
if (minSpace == 0) {
return true;
}
return minSpace < getFreeSpace();
}
private long getFreeSpace() throws IOException {
long biggestSeenSoFar = 0;
String[] localDirs = localStorage.getDirs();
for (int i = 0; i < localDirs.length; i++) {
DF df = null;
if (localDirsDf.containsKey(localDirs[i])) {
df = localDirsDf.get(localDirs[i]);
} else {
df = new DF(new File(localDirs[i]), fConf);
localDirsDf.put(localDirs[i], df);
}
long availOnThisVol = df.getAvailable();
if (availOnThisVol > biggestSeenSoFar) {
biggestSeenSoFar = availOnThisVol;
}
}
//Should ultimately hold back the space we expect running tasks to use but
//that estimate isn’t currently being passed down to the TaskTrackers
return biggestSeenSoFar;
}
判断方法是获取所有的 lcoalDir，计算出这些目录中可用空间最大一个目录的可用大小，为什么使用最大值作为可用大小，而不是所有目录可用空间总和，是因为 localDir 存放 task 的一些本地信息，这些信息是不能夸目录存放的，所以必须确保有一个目录能够容纳下所有的信息。当计算出 freeSpace 后，根据比较 localMinSpaceStart 值与 freeSpace 的大小决定是否接收新任务。
接下来就是获取 TT 的一些资源信息，如总虚拟内存，总物理内存，可用的虚拟内存，可用的物理内存，CPU 使用情况等。接着将这些值添加到 status 中去，发送给 JT。
5.TaskTracker.transmitHeartBeat：
//add node health information
TaskTrackerHealthStatus healthStatus = status.getHealthStatus();
synchronized (this) {
if (healthChecker != null) {
healthChecker.setHealthStatus(healthStatus);
} else {
healthStatus.setNodeHealthy(true);
healthStatus.setLastReported(0L);
healthStatus.setHealthReport(“”);
}
}
此处是检查 TT 的健康状况。
6.TaskTracker.transmitHeartBeat：
//
// Xmit the heartbeat
//
HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,
justStarted,
justInited,
askForNewTask,
heartbeatResponseId);
此处通过 RPC 调用 JT 的 heartbeat() 方法。传的参数包括：status——TT 自身的状态信息；justStarted——表示 TT 是否刚启动；justInited——表示 TT 是否刚初始化；askForNewTask——表示是否接收新任务；heartbeatResponseId——上次心跳返回的 responseId。方法的返回值是一个 HeartbeatResponse 对象，具体 JT 内的 heartbeat() 方法如何处理以及 HeartbeatResponse 内容会另外分析。继续往下走。
7.TaskTracker.transmitHeartBeat：
//
// The heartbeat got through successfully!
//
heartbeatResponseId = heartbeatResponse.getResponseId();
synchronized (this) {
for (TaskStatus taskStatus : status.getTaskReports()) {
if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
taskStatus.getRunState() != TaskStatus.State.UNASSIGNED &&
taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING &&
!taskStatus.inTaskCleanupPhase()) {
if (taskStatus.getIsMap()) {
mapTotal–;
} else {
reduceTotal–;
}
myInstrumentation.completeTask(taskStatus.getTaskID());
runningTasks.remove(taskStatus.getTaskID());
}
}
// Clear transient status information which should only
// be sent once to the JobTracker
for (TaskInProgress tip: runningTasks.values()) {
tip.getStatus().clearStatus();
}
}
// Force a rebuild of ‘status’ on the next iteration
status = null;
return heartbeatResponse;
首先从 HeartbeatResponse 返回值中获取 heartbeatResponseId。接下来对 TT 中的每个 TaskInProgress 的 status 信息进行判断，如果一个 task 处于 SUCCEEDED/FAILED/KILLED 状态，则表示该 task 已完成（不论是失败还是成功，亦或是被 kill 掉），如果该 task 是一个 map 任务，则 mapTotal 减一，该 task 是一个 reduce 任务，则 reduceTotal 减一，mapTotal/reduceTotal 记录当前 TT 所有处于运行状态（非 SUCCEEDED/FAILED/KILLED 状态）的 task 数量。
myInstrumentation.completeTask(taskStatus.getTaskID()) 此处将该 TT 所有完成任务数加一，runningTasks.remove(taskStatus.getTaskID()) 则是将该 task 从 runningTasks 队列中移除，所以可以知道 runningTasks 中只包含未完成的 task 信息。
接下来是清除 TaskInProgress 的 TaskStatus 的临时信息（diagnosticInfo），从 clearStatus() 方法的注释可以看出 diagnosticInfo 信息只是在 Task 向 TaskTracker，或者 TaskTracker 向 JobTracker 发送一个状态更新信息时的临时诊断信息，所以在发送完成之后需要清除。
到这里整个 TaskTracker 发送心跳信息的过程就完成了，方法返回值是 HeartbeatResponse 对象，即心跳的返回值。
更多 Hadoop 相关信息见 Hadoop 专题页面 http://www.linuxidc.com/topicnews.aspx?tid=13
在 Hadoop 中 JT（JobTracker）与 TT（TaskTracker）之间的通信是通过心跳机制完成的。JT 实现 InterTrackerProtocol 协议，该协议定义了 JT 与 TT 之间的通信机制——心跳。心跳机制实际上就是一个 RPC 请求，JT 作为 Server，而 TT 作为 Client，TT 通过 RPC 调用 JT 的 heartbeat 方法，将 TT 自身的一些状态信息发送给 JT，同时 JT 通过返回值返回对 TT 的指令。
目录 ：
Hadoop1.2.1 源码解析系列:JT 与 TT 之间的心跳通信机制——TT 篇 http://www.linuxidc.com/Linux/2014-06/103216.htm
Hadoop1.2.1 源码解析系列：JT 与 TT 之间的心跳通信机制——JT 篇 http://www.linuxidc.com/Linux/2014-06/103217.htm
Hadoop1.2.1 源码解析系列：JT 与 TT 之间的心跳通信机制——命令篇 http://www.linuxidc.com/Linux/2014-06/103218.htm
心跳有三个作用：
1）判断 TT 是否活着
2）报告 TT 的资源情况以及任务运行情况
3）为 TT 发送指令（如运行 task，kill task 等）
下面详细阅读下涉及到心跳调用的源码。
首先我们需要清楚，心跳机制是 TT 调用 JT 的方法，而非 JT 主动调用 TT 的方法。TT 通过 transmitHeartBeat 方法调用 JT 的 heartbeat 方法。
1.TaskTracker.transmitHeartBeat：
// Send Counters in the status once every COUNTER_UPDATE_INTERVAL
boolean sendCounters;
if (now > (previousUpdate + COUNTER_UPDATE_INTERVAL)) {
sendCounters = true;
previousUpdate = now;
}
else {
sendCounters = false;
}
根据 sendCounters 的间隔判断此次心跳是否发送计算器信息。
2.TaskTracker.transmitHeartBeat：
1.TaskTracker.transmitHeartBeat：
// Check if the last heartbeat got through…
// if so then build the heartbeat information for the JobTracker;
// else resend the previous status information.
//
if (status == null) {
synchronized (this) {
status = new TaskTrackerStatus(taskTrackerName, localHostname,
httpPort,
cloneAndResetRunningTaskStatuses(
sendCounters),
taskFailures,
localStorage.numFailures(),
maxMapSlots,
maxReduceSlots);
}
} else {
LOG.info(“Resending ‘status’ to ‘” + jobTrackAddr.getHostName() +
“‘ with reponseId ‘” + heartbeatResponseId);
}
此处根据 status 变量是否为 null，判断上次的心跳是否成功发送。tatus!=null，则表示上次的心跳尚未发送，所以直接将上次收集到的 TT 状态信息（封装在 status 中）发送给 JT；相反，status==null，则表示上次心跳已完成，重新收集 TT 的状态信息，同样封装到 status 中。下面详细看下 new TaskTrackerStatus（）方法。注意此处有个 cloneAndResetRunningTaskStatuses(sendCounters) 方法：
private synchronized List<TaskStatus> cloneAndResetRunningTaskStatuses(
boolean sendCounters) {
List<TaskStatus> result = new ArrayList<TaskStatus>(runningTasks.size());
for(TaskInProgress tip: runningTasks.values()) {
TaskStatus status = tip.getStatus();
status.setIncludeCounters(sendCounters);
// send counters for finished or failed tasks and commit pending tasks
if (status.getRunState() != TaskStatus.State.RUNNING) {
status.setIncludeCounters(true);
}
result.add((TaskStatus)status.clone());
status.clearStatus();
}
return result;
}
该方法中涉及到 runningTasks 队列，该队列保存了该 TT 上接收的所有未完成的 Task 任务，通过 runningTasks.values() 可以获取 TT 当前所有未完成的 Task，然后获取每个 TaskInProgress 的 status 信息，同时根据第一步判断出的 sendCounters（true/false）决定是否发送 counters 信息（includeCounters），即是否将 counters 对象序列化到 TaskStatus 对象中，这里需要注意如果 TaskInProgress 不处于 Running 状态，则 includeCounters 设为 true，即发送 counters 信息。
————————————– 分割线 ————————————–
Ubuntu 13.04 上搭建 Hadoop 环境 http://www.linuxidc.com/Linux/2013-06/86106.htm
Ubuntu 12.10 +Hadoop 1.2.1 版本集群配置 http://www.linuxidc.com/Linux/2013-09/90600.htm
Ubuntu 上搭建 Hadoop 环境（单机模式 + 伪分布模式）http://www.linuxidc.com/Linux/2013-01/77681.htm
Ubuntu 下 Hadoop 环境的配置 http://www.linuxidc.com/Linux/2012-11/74539.htm
单机版搭建 Hadoop 环境图文教程详解 http://www.linuxidc.com/Linux/2012-02/53927.htm
Hadoop LZO 安装教程 http://www.linuxidc.com/Linux/2013-01/78397.htm
Hadoop 集群上使用 Lzo 压缩 http://www.linuxidc.com/Linux/2012-05/60554.htm
————————————– 分割线 ————————————–
3.TaskTrackerStatus()：
public TaskTrackerStatus(String trackerName, String host,
int httpPort, List<TaskStatus> taskReports,
int taskFailures, int dirFailures,
int maxMapTasks, int maxReduceTasks) {
this.trackerName = trackerName;
this.host = host;
this.httpPort = httpPort;
this.taskReports = new ArrayList<TaskStatus>(taskReports);
this.taskFailures = taskFailures;
this.dirFailures = dirFailures;
this.maxMapTasks = maxMapTasks;
this.maxReduceTasks = maxReduceTasks;
this.resStatus = new ResourceStatus();
this.healthStatus = new TaskTrackerHealthStatus();
}
这里只是进行简单的变量复制操作，分析下其中一些参数的含义：
1）taskReports：包含该 TT 上目前所有的 Task 状态信息，其中的 counters 信息会根据之前判断 sendCounters 值进行决定是否发送，上一步有提到。
2）taskFailures：该 TT 上失败的 Task 总数（重启会清空），该参数帮助 JT 决定是否向该 TT 提交 Task，因为失败数越多表明该 TT 可能出现 Task 失败的概率越大。
3）dirFailures：这个值是 mapred.local.dir 参数设置的目录中有多少是不可用的（以后会详细提到）
4）maxMapSlots/maxReduceSlots：这个值是 TT 可使用的最大 map 和 reduce slot 数量
初始化完成，继续回到 TaskTracker.transmitHeartBeat 方法。
4.TaskTracker.transmitHeartBeat：
// Check if we should ask for a new Task
//
boolean askForNewTask;
long localMinSpaceStart;
synchronized (this) {
askForNewTask =
((status.countOccupiedMapSlots() < maxMapSlots ||
status.countOccupiedReduceSlots() < maxReduceSlots) &&
acceptNewTasks);
localMinSpaceStart = minSpaceStart;
}
if (askForNewTask) {
askForNewTask = enoughFreeSpace(localMinSpaceStart);
long freeDiskSpace = getFreeSpace();
long totVmem = getTotalVirtualMemoryOnTT();
long totPmem = getTotalPhysicalMemoryOnTT();
long availableVmem = getAvailableVirtualMemoryOnTT();
long availablePmem = getAvailablePhysicalMemoryOnTT();
long cumuCpuTime = getCumulativeCpuTimeOnTT();
long cpuFreq = getCpuFrequencyOnTT();
int numCpu = getNumProcessorsOnTT();
float cpuUsage = getCpuUsageOnTT();
status.getResourceStatus().setAvailableSpace(freeDiskSpace);
status.getResourceStatus().setTotalVirtualMemory(totVmem);
status.getResourceStatus().setTotalPhysicalMemory(totPmem);
status.getResourceStatus().setMapSlotMemorySizeOnTT(
mapSlotMemorySizeOnTT);
status.getResourceStatus().setReduceSlotMemorySizeOnTT(
reduceSlotSizeMemoryOnTT);
status.getResourceStatus().setAvailableVirtualMemory(availableVmem);
status.getResourceStatus().setAvailablePhysicalMemory(availablePmem);
status.getResourceStatus().setCumulativeCpuTime(cumuCpuTime);
status.getResourceStatus().setCpuFrequency(cpuFreq);
status.getResourceStatus().setNumProcessors(numCpu);
status.getResourceStatus().setCpuUsage(cpuUsage);
}
从源码中的注释可以知道，此处是 TT 根据自身资源使用情况判断是否接收 new task。
更多详情见请继续阅读下一页的精彩内容 ：http://www.linuxidc.com/Linux/2014-06/103216p2.htm
