TaskTracker也是作为一个单独的JVM来运行的，其main函数就是TaskTracker的入口函数，当运行start-all.sh时，脚本就是通过SSH运行该函数来启动TaskTracker的。
TaskTracker是JobTracker和Task之间的桥梁：一方面，从JobTracker接收并执行各种命令：运行任务、提交任务、杀死任务等；另一方面，将本地节点上各个任务的状态通过心跳周期性汇报给JobTracker。TaskTracker与JobTracker和Task之间采用了RPC协议进行通信。
public void run() {
try {
getUserLogManager().start();
startCleanupThreads();
boolean denied = false;
while (running && !shuttingDown && !denied) {
boolean staleState = false;
try {
// This while-loop attempts reconnects if we get network errors
while (running && !staleState && !shuttingDown && !denied) { //如果连接断开，则会循环尝试连接JobTracker
try {
//如果连接JobTracker服务成功，TaskTracker就会调用offerService()函数进入主执行循环中。
//这个循环会每隔一段时间与JobTracker通讯一次，调用transmitHeartBeat()，获得HeartbeatResponse信息。
State osState = offerService();
if (osState == State.STALE) {
staleState = true;
} else if (osState == State.DENIED) {
denied = true;
}
} catch (Exception ex) {
if (!shuttingDown) {
LOG.info("Lost connection to JobTracker [" +
jobTrackAddr + "]. Retrying...", ex);
try {
Thread.sleep(5000);
} catch (InterruptedException ie) {
}
}
}
}
} finally {
close();
}
if (shuttingDown) { return; }
LOG.warn("Reinitializing local state");
initialize(); //初始化所有成员和参数
}
if (denied) {
shutdown();
}
} catch (IOException iex) {
LOG.error("Got fatal exception while reinitializing TaskTracker: " +
StringUtils.stringifyException(iex));
return;
}
catch (InterruptedException i) {
LOG.error("Got interrupted while reinitializing TaskTracker: " +
i.getMessage());
return;
}
}getUserLogManager().start()会启动两个线程：用户日志清理线程和监控日志行为线程；会启动一个taskCleanupThread线程，这个线程会始终监视BlockingQueue<TaskTrackerAction> tasksToCleanup(用来存放要被清理的TaskInProgress)，里面存储的是KillJobAction或者是KillTaskAction类型；接下来是两层的while循环，其中内层循环是一旦连接断开就循环尝试连接JobTracker，循环主体是State osState = offerService()，offerService()方法一旦执行如果没有其它情况将会一直执行，如果内层while循环退出则尝试重新initialize()，如果offerService()方法返回的状态是DENIED则退出所有循环并执行shutdown()。
/**
* Main service loop. Will stay in this loop forever.
*/
State offerService() throws Exception {
long lastHeartbeat = System.currentTimeMillis();//上一次发心跳距现在时间
//此循环主要根据控制完成task个数控制心跳间隔。//TaskTracker进行是一直存在的
while (running && !shuttingDown) {
try {
long now = System.currentTimeMillis();//获得当前时间 // accelerate to account for multiple finished tasks up-front
//通过完成的任务数动态控制心跳间隔时间
long remaining =
(lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;
while (remaining > 0) {
// sleeps for the wait time or
// until there are *enough* empty slots to schedule tasks
synchronized (finishedCount) {
finishedCount.wait(remaining); // Recompute
now = System.currentTimeMillis();
remaining =
(lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now; if (remaining <= 0) {
// Reset count
finishedCount.set(0);
break;
}
}
} // If the TaskTracker is just starting up:
// 1. Verify the buildVersion
// 2. Get the system directory & filesystem
if(justInited) {
String jobTrackerBV = jobClient.getBuildVersion();
if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {
String msg = "Shutting down. Incompatible buildVersion." +
"\nJobTracker's: " + jobTrackerBV +
"\nTaskTracker's: "+ VersionInfo.getBuildVersion();
LOG.error(msg);
try {
jobClient.reportTaskTrackerError(taskTrackerName, null, msg);
} catch(Exception e ) {
LOG.info("Problem reporting to jobtracker: " + e);
}
return State.DENIED;
} String dir = jobClient.getSystemDir();
if (dir == null) {
throw new IOException("Failed to get system directory");
}
systemDirectory = new Path(dir);
systemFS = systemDirectory.getFileSystem(fConf);
} now = System.currentTimeMillis();
if (now > (lastCheckDirsTime + diskHealthCheckInterval)) {
localStorage.checkDirs();
lastCheckDirsTime = now;
int numFailures = localStorage.numFailures();
// Re-init the task tracker if there were any new failures
if (numFailures > lastNumFailures) {
lastNumFailures = numFailures;
return State.STALE;
}
} // Send the heartbeat and process the jobtracker's directives
//在transmitHeartBeat()函数处理中，
//TaskTracker会创建一个新的TaskTrackerStatus对象记录目前任务的执行状况，
//检查目前执行的Task数目以及本地磁盘的空间使用情况等，
//如果可以接收新的Task则设置heartbeat()的askForNewTask参数为true。
//然后通过IPC接口调用JobTracker的heartbeat()方法发送过去，
//heartbeat()返回值TaskTrackerAction数组。
HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);//发送Heartbeat到JobTracker，得到response // Note the time when the heartbeat returned, use this to decide when to send the
// next heartbeat
lastHeartbeat = System.currentTimeMillis(); // Check if the map-event list needs purging
Set<JobID> jobs = heartbeatResponse.getRecoveredJobs();
if (jobs.size() > 0) {
synchronized (this) {
// purge the local map events list
for (JobID job : jobs) {
RunningJob rjob;
synchronized (runningJobs) {
rjob = runningJobs.get(job);
if (rjob != null) {
synchronized (rjob) {
FetchStatus f = rjob.getFetchStatus();
if (f != null) {
f.reset();
}
}
}
}
} // Mark the reducers in shuffle for rollback
synchronized (shouldReset) {
for (Map.Entry<TaskAttemptID, TaskInProgress> entry
: runningTasks.entrySet()) {
if (entry.getValue().getStatus().getPhase() == Phase.SHUFFLE) {
this.shouldReset.add(entry.getKey());
}
}
}
}
}
//然后调用HeartbeatResponse的getActions()函数获得JobTracker传过来的所有指令即一个TaskTrackerAction数组。
TaskTrackerAction[] actions = heartbeatResponse.getActions();
if(LOG.isDebugEnabled()) {
LOG.debug("Got heartbeatResponse from JobTracker with responseId: " +
heartbeatResponse.getResponseId() + " and " +
((actions != null) ? actions.length : 0) + " actions");
}
if (reinitTaskTracker(actions)) {
return State.STALE;
} // resetting heartbeat interval from the response.
heartbeatInterval = heartbeatResponse.getHeartbeatInterval();
justStarted = false;
justInited = false;
if (actions != null){
//遍历这个数组，如果是一个新任务指令即LaunchTaskAction则调用调用addToTaskQueue加入到待执行队列，
//否则加入到tasksToCleanup队列，交给一个taskCleanupThread线程来处理，如执行KillJobAction或者KillTaskAction等。
for(TaskTrackerAction action: actions) {
if (action instanceof LaunchTaskAction) {
addToTaskQueue((LaunchTaskAction)action); //如果是运行一个新的Task，则将Action添加到任务队列中,加入TaskLauncher线程的执行队列
} else if (action instanceof CommitTaskAction) {
CommitTaskAction commitAction = (CommitTaskAction)action;
if (!commitResponses.contains(commitAction.getTaskID())) {
LOG.info("Received commit task action for " +
commitAction.getTaskID());
commitResponses.add(commitAction.getTaskID());
}
} else {//杀死任务或作业
tasksToCleanup.put(action);
}
}
}
//杀死一定时间没没有汇报进度的task
markUnresponsiveTasks();
//当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时，寻找合适的任务将其杀掉以释放空间
killOverflowingTasks(); //we've cleaned up, resume normal operation
if (!acceptNewTasks && isIdle()) {
acceptNewTasks=true;
}
//The check below may not be required every iteration but we are
//erring on the side of caution here. We have seen many cases where
//the call to jetty's getLocalPort() returns different values at
//different times. Being a real paranoid here.
checkJettyPort(server.getPort());
} catch (InterruptedException ie) {
LOG.info("Interrupted. Closing down.");
return State.INTERRUPTED;
} catch (DiskErrorException de) {
String msg = "Exiting task tracker for disk error:\n" +
StringUtils.stringifyException(de);
LOG.error(msg);
synchronized (this) {
jobClient.reportTaskTrackerError(taskTrackerName,
"DiskErrorException", msg);
}
return State.STALE;
} catch (RemoteException re) {
String reClass = re.getClassName();
if (DisallowedTaskTrackerException.class.getName().equals(reClass)) {
LOG.info("Tasktracker disallowed by JobTracker.");
return State.DENIED;
}
} catch (Exception except) {
String msg = "Caught exception: " +
StringUtils.stringifyException(except);
LOG.error(msg);
}
} return State.NORMAL;
}
方法中有一个循环，没有其他状况会一直循环执行，循环内首先会阻塞心跳间隔时间，心跳间隔是动态的会不断修正的；如果TaskTracker是刚刚启动，需要先确认版本一致否则直接返回State.DENIED，然后获取system directory和filesystem；超过磁盘检查间隔就对磁盘进行检查，是否有损坏，然后检查localStorage.numFailures()，如果大于lastNumFailures，则直接返回State.STALE，对TaskTracker重新初始化； transmitHeartBeat(now)方法会发送心跳到JobTracker，并返回心跳响应信息heartbeatResponse，相应信息包括两部分，一个是作业集合recoveredJobs它是上次关闭JobTracker时正在运行的作业集合，重启JobTracker后需要恢复这些作业的运行状态(前提是用户启用了作业恢复功能)，而TaskTracker收到该作业集合后，需重置这些作业对应Reduce Task的FetchStatus信息，从而迫使这些Reduce Task重新从Map Task端拷贝数据，另一部分是需要执行的命令列表；heartbeatResponse.getActions()获取JobTracker发送过来的指令数组，首先要检查的是是否有重新初始化TaskTracker的action，如果有则返回State.STALE，会重新初始化TaskTracker；修正心跳间隔；遍历actions数组，根据action的类型加入不同的数据结构中，是LaunchTaskAction：如果是action.getTask().isMapTask()==true，则将action加入mapLauncher，否则加入reduceLauncher，这两个launcher都是TaskLauncher（该类是TaskTracker类的一个内部类，具有一个数据成员，是 TaskTracker.TaskInProgress类型的队列。在此特别注意，在TaskTracker类内部所提到的TaskInProgress 类均为TaskTracker的内部类，我们用TaskTracker.TaskInProgress表示，一定要和mapred包中的 TaskInProgress类区分，后者我们直接用TaskInProgress表示）；是CommitTaskAction：则放入commitResponses，commitResponses.add(commitAction.getTaskID())；其他的类型放入tasksToCleanup.put(action)，表示要清理，如执行KillJobAction或者KillTaskAction等。markUnresponsiveTasks()方法遍历runningTasks杀死一定时间没没有汇报进度的task(purgeTask(tip, true)方法)。killOverflowingTasks()方法，当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时，寻找合适的任务将其杀掉以释放空间，期间不接受任何新的task，acceptNewTasks=false，通过findTaskToKill(null)方法(会遍历runningTasks，优先考虑杀死reduce任务)找到合适的TaskInProgress killMe，执行purgeTask(killMe, false)。一旦空间不足而出现杀死task的情况出现，就会一直不接受任何新的task，直到所有的task执行完毕和所有的清理task也执行完毕，但仍会正常向JobTracker发送心跳信息，信息内容就会有所变化。
transmitHeartBeat(now)方法是构造并将心跳信息发送到JobTracker。主要是构造一个TaskTrackerStatus这是要发送的东西，其内容包括任务的运行状态信息、TaskTracker资源信息、健康状态信息。如果可以接收新的Task则设置askForNewTask参数为true。当满足下面的条件的时候，此TaskTracker请求JobTracker为其分配一个新的Task来运行：当前TaskTracker正在运行的map task的个数小于可以运行的map task的最大个数；当前TaskTracker正在运行的reduce task的个数小于可以运行的reduce task的最大个数；acceptNewTasks==true。如果askForNewTask==true则对TaskTrackerStatus的实例status进行一些设置。然后对healthStatus = status.getHealthStatus()中的healthStatus进行一些设置。然后向JobTracker发送心跳并接受相应信息HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, justStarted,justInited,askForNewTask, heartbeatResponseId)这是一个RPC调用，具体可以查看JobTracker.heartbeat方法，后续再讲。调用成功后更新heartbeatResponseId。遍历所有task的状态TaskStatus，对那些状态为SUCCEEDED或者FAILED或者KILLED，做一些统计信息，根据task类型使得mapTotal或者reduceTotal减一，runningTasks去除这个task，然后清空runningTasks中所有的TaskInProgress.getStatus().clearStatus()状态信息，这些状态信息是瞬时的，仅发送一次，status = null，这些瞬时的状态信息是在构造TaskTrackerStatus时通过cloneAndResetRunningTaskStatuses(sendCounters)生成的。最后返回心跳结果heartbeatResponse。
这样offerService()方法通过while循环一直阻塞一定的心跳间隔，然后获取JobTracker的心跳应答信息，根据其中的action添加到不同的数据结构中，并做一些检查控制TaskTracker能够较为合理的运行，总是一遍又一遍的做这些。
至此，TaskTracker的启动过程讲解完了，错误之处还请指正。