/**

* TaskTracker是一个独立的服务，所以有一个main方法。

*/

public static void main(String argv[]) throws Exception {

    ...... 打印信息，以及检查argv参数

    try {

      JobConf conf=new JobConf();

      ...... JMX信息设置

      TaskTracker tt = new TaskTracker(conf); // 创建TaskTracker对象

      MBeans.register("TaskTracker", "TaskTrackerInfo", tt); // 注册MXBean

      tt.run(); // 执行

    } catch (Throwable e) {

      ..... 异常处理

    }

  }

........

FILE_CACHE_SIZE = conf.getInt("mapred.tasktracker.file.cache.size", 2000); // 获取文件cache大小，默认2000

maxMapSlots = conf.getInt("mapred.tasktracker.map.tasks.maximum", 2); // 获取配置的map slot大小，默认2

maxReduceSlots = conf.getInt("mapred.tasktracker.reduce.tasks.maximum", 2); // 获取配置的reduce slot个数，默认为2

diskHealthCheckInterval = conf.getLong(DISK_HEALTH_CHECK_INTERVAL_PROPERTY, DEFAULT_DISK_HEALTH_CHECK_INTERVAL); // 获取节点健康检查间隔时间，默认为60s

..........

taskController = (TaskController) ReflectionUtils.newInstance(taskControllerClass, fConf); // 创建控制器对象

..........

initialize(); // 初始化

.........

server.start(); // 启动httpserver

this.httpPort = server.getPort();

checkJettyPort(httpPort); // 检查jetty服务器，由于当端口号为小于0的情况下，jetty有bug，所以在这种情况下，设置shuttingDown为true。

// RPC initialization

int max = maxMapSlots > maxReduceSlots ? maxMapSlots : maxReduceSlots; // RPC初始值，采用最大的slot值。

//set the num handlers to max*2 since canCommit may wait for the duration of a heartbeat RPC

// 创建和Task通信的Server端对象，线程大小为2倍的slot大小。

this.taskReportServer = RPC.getServer(this, bindAddress, tmpPort, 2 * max, false, this.fConf, this.jobTokenSecretManager);

this.taskReportServer.start(); // 启动

........

// Initialize DistributedCache

this.distributedCacheManager = new TrackerDistributedCacheManager(this.fConf, taskController); // 初始化分布式缓存系统

this.distributedCacheManager.startCleanupThread(); // 启动

this.jobClient = (InterTrackerProtocol)UserGroupInformation.getLoginUser().doAs(

　　　　new PrivilegedExceptionAction<Object>() {

　　　　　　　　public Object run() throws IOException {

　　　　　　　　　　　　return RPC.waitForProxy(InterTrackerProtocol.class,InterTrackerProtocol.versionID,jobTrackAddr, fConf);

　　　　　　　　}

　　　　}); // 创建和JobTracker通信的客户端对象

......

// 创建获取map的线程

this.mapEventsFetcher = new MapEventsFetcherThread();

mapEventsFetcher.setDaemon(true);

mapEventsFetcher.setName("Map-events fetcher for all reduce tasks " + "on " + taskTrackerName);

mapEventsFetcher.start();

.............

// 创建map/reduce任务启动线程并run

mapLauncher = new TaskLauncher(TaskType.MAP, maxMapSlots);

reduceLauncher = new TaskLauncher(TaskType.REDUCE, maxReduceSlots);

mapLauncher.start();

reduceLauncher.start();

boolean denied = false; // 是否JobTracker拒绝TaskTracker的心跳报告

while (running && !shuttingDown && !denied) { // 如果当然TaskTracker处于运行状态，而且jetty服务器运行正常，并且没有被JobTracker拒绝请求，那么就一直循环

    boolean staleState = false; // 状态信息

    try {

      // This while-loop attempts reconnects if we get network errors

      while (running && !staleState && !shuttingDown && !denied) { // 外面的条件加上如果状态信息为false，那么就一直循环，这种方式是为了保证，当网络出现问题的时候，能够重新连接。

        try {

          State osState = offerService(); // 进行具体和JobTracker的通信

          if (osState == State.STALE) { // 返回的状态不正常

            staleState = true;

          } else if (osState == State.DENIED) { // 返回的状态不正常

            denied = true;

          }

        } catch (Exception ex) {

          // 异常处理

        }

      }

    } finally {

      close(); // 只要是出现异常，或者是退出死循环，那么都关闭资源，重新启动资源

    }

    if (shuttingDown) { return; } // 如果是jetty服务器问题，那么直接退出

    initialize(); // 网络出现问题，重新连接

}

if (denied) { // 如果JobTracker拒绝请求，那么关闭httpserver

    shutdown();

}

while (running && !shuttingDown) {// Send the heartbeat and process the jobtracker's directives

        // 发送心跳报告，并获取任务命令

        HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);

        if (actions != null){  // 如果命令不为空，那就处理命令

          for(TaskTrackerAction action: actions) {

            if (action instanceof LaunchTaskAction) { // 如果命令为启动task命令

              addToTaskQueue((LaunchTaskAction)action);

            } else if (action instanceof CommitTaskAction) { // 提交task命令

             ...

            } else {

              ....

            }

          }

        }

        markUnresponsiveTasks(); // 杀死一定时间内无反应的任务

        killOverflowingTasks(); // 当磁盘空间少于mapred.local.dir.minspacekill时候，寻找合适的任务将其杀死以释放空间。