zoukankan      html  css  js  c++  java
  • 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);
                }
              }
            }
    } 
  • 相关阅读:
    python初学第一节课
    关于float类型和u32类型问题
    今天的工作状态,规划未来一段时间内必须完成的事情(Record the working status of today,planning for the next period of time must be completed)
    STM32 硬件I2C初始化 I2C1_GPIO_AF_Config
    C语言编程规范--------12 宏
    C语言编程规范--------11 代码测试、维护
    C语言编程规范--------10 代码编辑、编译、审查
    C语言编程规范--------9 质量保证
    C语言编程规范--------8 效率
    C语言编程规范--------7 可测性
  • 原文地址:https://www.cnblogs.com/yueliming/p/3009869.html
Copyright © 2011-2022 走看看