TaskTracker发送Heartbeat以及接受HeartbeatResponse

zliyt

　　          我们通过在命令行输入start-all.sh启动hadoop服务，该脚本通过SSH运行各个节点上的TaskTracker类的main()来启动TaskTracker，它是作为一个单独的JVM来运行的。TaskTracker类实现了MapReduce模型中的TaskTracker的功能。TaskTracker的main函数如下：

    main（）{
    TaskTracker tt = new  TaskTracker();
        tt. run();
.......　　}
　　TaskTracker的run()方法主要调用了offerService()，其主要代码如下：


while (running && !shuttingDown) {//TaskTracker运行期间一直执行，每隔一段时间就想JobTracker发送一次心跳包，并接受JobTracker发送的HeartbeatResponse
      try {
      //判断是否该发送下一个心跳包
        while (remaining > 0) {
...
        }
        // TaskTracker 刚启动:
        // 1. 验证buildVersion
        // 2. 为TaskTracker创建一个System Directory
        if(justInited) {
          String jobTrackerBV = jobClient.getBuildVersion();
        ...
          systemDirectory = new Path(dir);
          systemFS = systemDirectory.getFileSystem(fConf);
        }
//每个一段时间检查一下TaskTrack的工作目录
        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;
          }
        }
         //发送心跳包，并向JobTracker请求Task
        HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);
...
//从心跳包的响应中获取它要执行的任务

        TaskTrackerAction[] actions = heartbeatResponse.getActions();


//将相应的Task放入到特定的Queue

if (actions != null){
          for(TaskTrackerAction action: actions) {
//将要运行一个Task
            if (action instanceof LaunchTaskAction) {
              addToTaskQueue((LaunchTaskAction)action);
            } 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 {//待清理的Task
              tasksToCleanup.put(action);
            }
          }
        }
}