hadoop 源码分析(四)JobTracker 添加job 到schduler 队列中

  启动 JobTracker
1. 进入main方法：
   

 //执行startTracker 方法
      JobTracker tracker = startTracker(new JobConf());
     //执行offerService()方法
      tracker.offerService();

3.startTracker()方法中
       

 // new JobTracker 方法
          result = new JobTracker(conf, identifier);
        result.taskScheduler.setTaskTrackerManager(result);

2. JobTracker()构造方法中初始化信息
     (1) static constants 变量
     (2) 加载调度器 默认为FIFO 调度
      /

/ Create the scheduler
          Class<? extends TaskScheduler> schedulerClass
          = conf.getClass("mapred.jobtracker.taskScheduler",
          JobQueueTaskScheduler.class, TaskScheduler.class);
taskScheduler=(TaskScheduler)ReflectionUtils.newInstance(schedulerClass, conf);

taskScheduler 默认的执行类为JobQueueTaskScheduler ,当启动JobTracker 的时候 调用了 tracker.offerService();该方法执行了父类的 start()方法.该start()方法为 JobQueueTaskScheduler 的start方法:
       

  @Override
          public synchronized void start() throws IOException {
           super.start();
taskTrackerManager.addJobInProgressListener(jobQueueJobInProgressListener);
eagerTaskInitializationListener.setTaskTrackerManager(taskTrackerManager);
           eagerTaskInitializationListener.start();
           taskTrackerManager.addJobInProgressListener(
           eagerTaskInitializationListener);
  }

在这个方法中调用了addJobInProgressListener()将lister 加入到了 jobTracker中,这个过程很绕,可通过下面的流程图梳理清楚

3. 启动jettyServer

infoServer.addServlet("reducegraph", "/taskgraph", TaskGraphServlet.class);
infoServer.start();

  JobTracker 提交job
1. jobClient()通过代理 调用JobTracker的submit方法提交job
2. submitJob方法中 调用了 addJob()将job添加到job队列中，等待执行
3. addJob方法：
   

 private synchronized JobStatus addJob(JobID jobId, JobInProgress job) 
     throws IOException {
    totalSubmissions++;

    synchronized (jobs) {
      synchronized (taskScheduler) {
        jobs.put(job.getProfile().getJobID(), job);
       // jobInProgressListeners list 在start jobTracker 的时候,在JobQueueTaskScheduler 的start方法中初始化加入了两个listener ：
       eagerTaskInitializationListener 和jobQueueJobInProgressListener 
        for (JobInProgressListener listener : jobInProgressListeners) {
          listener.jobAdded(job);
        }
      }
    }
    myInstrumentation.submitJob(job.getJobConf(), jobId);
    job.getQueueMetrics().submitJob(job.getJobConf(), jobId);

    LOG.info("Job " + jobId + " added successfully for user '" 
             + job.getJobConf().getUser() + "' to queue '" 
             + job.getJobConf().getQueueName() + "'");
    AuditLogger.logSuccess(job.getUser(), 
        Operation.SUBMIT_JOB.name(), jobId.toString());
    return job.getStatus();
  }