TaskTracker发送Heartbeat以及接受HeartbeatResponse

          我们通过在命令行输入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);
            }
          }
        }
} 
原文地址:https://www.cnblogs.com/yueliming/p/3009869.html