MapReduce job在JobTracker初始化的源码级分析

mapreduce job提交流程源码级分析（三）中已经说明用户最终调用JobTracker.submitJob方法来向JobTracker提交作业。而这个方法的核心提交方法是JobTracker.addJob(JobID jobId, JobInProgress job)方法，这个addJob方法会把Job提交到调度器(默认是JobQueueTaskScheduler)的监听器JobQueueJobInProgressListener和EagerTaskInitializationListener(本文只讨论默认调度器)中，使用方法jobAdded(JobInProgress job)，JobQueueJobInProgressListener任务是监控各个JobInProcess生命周期中的变化；EagerTaskInitializationListener是发现有新Job后对其初始化的。

一、JobQueueJobInProgressListener.jobAdded(JobInProgress job)方法。就一句代码jobQueue.put(new JobSchedulingInfo(job.getStatus()), job)，先构建一个JobSchedulingInfo对象，然后和JobInProgress对应起来放入jobQueue中。JobSchedulingInfo类维护这调度这个job必备的一些信息，比如优先级(默认是NORMAL)、JobID以及开始时间startTime。

二、EagerTaskInitializationListener.jobAdded(JobInProgress job)方法。　

/**
   * We add the JIP to the jobInitQueue, which is processed
   * asynchronously to handle split-computation and build up
   * the right TaskTracker/Block mapping.
   */
  @Override
  public void jobAdded(JobInProgress job) {
    synchronized (jobInitQueue) {
      jobInitQueue.add(job);　　//添加进List<JobInProgress> jobInitQueue
      resortInitQueue();
      jobInitQueue.notifyAll();　　//唤醒阻塞的进程
    }

  }

上面方法中resortInitQueue()方法主要是对jobInitQueue中JobInProcess进行排序，先按照优先级排序，相同的再按开始时间。EagerTaskInitializationListener.start()在调度器初始化时JobQueueTaskScheduler.start()就调用了，所以先于jobAdded方法调用。EagerTaskInitializationListener.start()代码如下：

1 public void start() throws IOException {
2     this.jobInitManagerThread = new Thread(jobInitManager, "jobInitManager");
3     jobInitManagerThread.setDaemon(true);
4     this.jobInitManagerThread.start();
5   }

start()方法会启动一个线程：JobInitManager。

/////////////////////////////////////////////////////////////////
  //  Used to init new jobs that have just been created
  /////////////////////////////////////////////////////////////////
  class JobInitManager implements Runnable {

    public void run() {
      JobInProgress job = null;
      while (true) {
        try {
          synchronized (jobInitQueue) {
            while (jobInitQueue.isEmpty()) {
              jobInitQueue.wait();
            }
            job = jobInitQueue.remove(0);
          }
          threadPool.execute(new InitJob(job));
        } catch (InterruptedException t) {
          LOG.info("JobInitManagerThread interrupted.");
          break;
        }
      }
      LOG.info("Shutting down thread pool");
      threadPool.shutdownNow();
    }
  }

  class InitJob implements Runnable {

    private JobInProgress job;

    public InitJob(JobInProgress job) {
      this.job = job;
    }

    public void run() {
      ttm.initJob(job);//对应JobTracker的对应方法
    }
  }

返回栏目页：http://www.bianceng.cnhttp://www.bianceng.cn/webkf/tools/

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索方法
， job
， start
， 调度
， jobtracker
， mapreduce 多线程
调用同一个job
mapreduce jobtracker、mapreduce job、mapreduce.job.maps、mapreduce多个job、mapreduce kill job，以便于您获取更多的相关知识。