TaskTracker启动过程源码级分析

　　TaskTracker也是作为一个单独的JVM来运行的，其main函数就是TaskTracker的入口函数，当运行start-all.sh时，脚本就是通过SSH运行该函数来启动TaskTracker的。

　　TaskTracker是JobTracker和Task之间的桥梁：一方面，从JobTracker接收并执行各种命令：运行任务、提交任务、杀死任务等；另一方面，将本地节点上各个任务的状态通过心跳周期性汇报给JobTracker。TaskTracker与JobTracker和Task之间采用了RPC协议进行通信。

　　TaskTracker的功能：1.汇报心跳：Tracker周期性将所有节点上各种信息通过心跳机制汇报给JobTracker。这些信息包括两部分：机器级别信息：节点健康情况、资源使用情况等；任务级别信息：任务执行进度、任务运行状态等。2.执行命令：JobTracker会给TaskTracker下达各种命令，主要包括：启动任务(LaunchTaskAction)、提交任务(CommitTaskAction)、杀死任务(KillTaskAction)、杀死作业(KillJobAction)和重新初始化(TaskTrackerReinitAction)。

　　main()方法中主要工作就是初始化TaskTracker并启动它。

　　一、初始化：TaskTracker tt = new TaskTracker(conf)，TaskTracker(conf)获取本机的一些配置信息，初始化服务器并启动服务器（StatusHttpServer）。

　　获取map和reduce最大的slot数量；获取jobTracker的地址jobTrackAddr；构造TaskTracker的HttpServer，可以从浏览器查看该tasktracker的运行情况及所有task；获取从HDFS下载文件的本地目录localStorage(即"mapred.local.dir")；构造一个TaskController（是一个接口）默认是DefaultTaskController，用来控制任务的初始化、终结、清理tasks的，也可以启动和杀死tasks的JVM；调用initialize()方法，该方法是真正的构造函的地方，使用单独的方法可以再次调用；启动HttpServer。

　　initialize()方法会清理一些历史文件(第一次启动TaskTracker不会有历史文件)，并新建一些目录；初始化一些参数；新建一个JvmManager来管理着mapJvmManager和reduceJvmManager，这俩都是JvmManagerForType，JvmManager.launchJvm(TaskRunner t, JvmEnv env)方法会启动Task JVM运行Task，这个过程还比较复杂后续再说；构造taskReportServer，Task通过RPC向该Server汇报进度；构造一个distributedCacheManager分布式缓存管理并启动CleanupThread线程周期性的检查和清理不再使用的文件；然后实例化InterTrackerProtocol的jobClient，InterTrackerProtocol是用于TaskTracker和JobTracker通信的协议；启动一个MapEventsFetcherThread线程用于获取Map任务的输出数据信息，获取已运行完成的Map Task列表，为Reduce Task远程拷贝数据做准备；初始化内存，做任务内存监控；构造两个TaskLauncher线程：mapLauncher和reduceLauncher并启动，这个线程的run函数就是不断监测taskToLaunch队列中是否有新的 TaskTracker.TaskInProgress对象加入。如果有则从中取出一个对象，然后调用TaskTracker类的 startNewTask(TaskInProgress tip)来启动一个task。

　　二、启动：tt.run()启动TaskTracker线程，维护一个和JobTracker通信的链接，其中run函数主要调用了offerService函数。JobTracker从不会主动与TaskTracker通信，总是被动等待TaskTracker汇报信息并领取其对应的命令。　　

  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()。

 　　offerService()方法代码较多，如下：

/**
   * 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." +
            "
JobTracker's: " + jobTrackerBV + 
            "
TaskTracker'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:
" +
          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的启动过程讲解完了，错误之处还请指正。

　　参考：1、http://www.2cto.com/kf/201311/260826.html

　　　　　2、董西成，《hadoop技术内幕---深入理解MapReduce架构设计与实现原理》