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();
}