Hadoop源码之JobTracker

nihaogirl

　　转自:http://blog.csdn.net/wwtang9527/article/details/8330472
　　JobTracker是Map/Reducer中任务调度的服务器。
　　1、有如下线程为其服务：
　　1）提供两组RPC服务（InterTrackerProtocol、JobSubmissionProtocol）的1个Listener线程与默认10个Handler线程；
　　2）提供任务执行情况查询的一组web服务线程，包括Socker Listener等；
　　3）ExpireTrackers：用来停止已经无效的TaskTracker服务；

[java]
view plaincopyprint?

• 
  synchronized(taskTrackers){
  
• 
  synchronized(trackerExpiryQueue){
  
• 
  longnow=System.currentTimeMillis();
  
• TaskTrackerStatusleastRecent=null;
  
• 
  while((trackerExpiryQueue.size()>0)&&
  
• ((leastRecent=(TaskTrackerStatus)trackerExpiryQueue.first())!=null)&&
  
• (now-leastRecent.getLastSeen()>TASKTRACKER_EXPIRY_INTERVAL)){
  
• 
  
• //Removeprofilefromheadofqueue
  
  
• trackerExpiryQueue.remove(leastRecent);
  
• StringtrackerName=leastRecent.getTrackerName();
  
• 
  
• //Figureoutiflast-seentimeshouldbeupdated,oriftrackerisdead
  
  
• TaskTrackerStatusnewProfile=(TaskTrackerStatus)taskTrackers.get(leastRecent.getTrackerName());
  
• //ItemsmightleavethetaskTrackersetthroughothermeans;the
  
  
• //statusstoredin'taskTrackers'mightbenull,whichmeansthe
  
  
• //trackerhasalreadybeendestroyed.
  
  
• 
  if(newProfile!=null){
  
• 
  if(now-newProfile.getLastSeen()>TASKTRACKER_EXPIRY_INTERVAL){
  
• //Removecompletely
  
• updateTaskTrackerStatus(trackerName,null);
  
• lostTaskTracker(leastRecent.getTrackerName());
  
• }else{
  
• 
  //Updatetimebyinsertinglatestprofile
  
• trackerExpiryQueue.add(newProfile);
  
• }
  
• }
  
• }
  
• }
  
• }
  

synchronized (taskTrackers) {
synchronized (trackerExpiryQueue) {
long now = System.currentTimeMillis();
TaskTrackerStatus leastRecent = null;
while ((trackerExpiryQueue.size() > 0) &&
((leastRecent = (TaskTrackerStatus) trackerExpiryQueue.first()) != null) &&
(now - leastRecent.getLastSeen() > TASKTRACKER_EXPIRY_INTERVAL)) {
// Remove profile from head of queue
trackerExpiryQueue.remove(leastRecent);
String trackerName = leastRecent.getTrackerName();
// Figure out if last-seen time should be updated, or if tracker is dead
TaskTrackerStatus newProfile = (TaskTrackerStatus) taskTrackers.get(leastRecent.getTrackerName());
// Items might leave the taskTracker set through other means; the
// status stored in 'taskTrackers' might be null, which means the
// tracker has already been destroyed.
if (newProfile != null) {
if (now - newProfile.getLastSeen() > TASKTRACKER_EXPIRY_INTERVAL) {
// Remove completely
updateTaskTrackerStatus(trackerName, null);
lostTaskTracker(leastRecent.getTrackerName());
} else {
// Update time by inserting latest profile
trackerExpiryQueue.add(newProfile);
}
}
}
}
}
　　4）RetireJobs：用来除去已经完成任务的TaskTracker；

[java]
view plaincopyprint?

• 
  synchronized(jobs){
  
• 
  synchronized(jobInitQueue){
  
• 
  synchronized(jobsByArrival){
  
• 
  for(Iteratorit=jobs.keySet().iterator();it.hasNext();){
  
• Stringjobid=(String)it.next();
  
• JobInProgressjob=(JobInProgress)jobs.get(jobid);
  
  
• 
  if(job.getStatus().getRunState()!=JobStatus.RUNNING&&
  
• job.getStatus().getRunState()!=JobStatus.PREP&&
  
• (job.getFinishTime()+RETIRE_JOB_INTERVAL<System.currentTimeMillis())){
  
• it.remove();
  
• 
  
• jobInitQueue.remove(job);
  
• jobsByArrival.remove(job);
  
• }
  
• }
  
• }
  
• }
  
• }
  

synchronized (jobs) {
synchronized (jobInitQueue) {
synchronized (jobsByArrival) {
for (Iterator it = jobs.keySet().iterator(); it.hasNext(); ) {
String jobid = (String) it.next();
JobInProgress job = (JobInProgress) jobs.get(jobid);
if (job.getStatus().getRunState() != JobStatus.RUNNING &&
job.getStatus().getRunState() != JobStatus.PREP &&
(job.getFinishTime() + RETIRE_JOB_INTERVAL < System.currentTimeMillis())) {
it.remove();
jobInitQueue.remove(job);
jobsByArrival.remove(job);
}
}
}
}
}
　　

5）JobInitThread：用来对job做一些初始化的工作；

[java]
view plaincopyprint?

• 
  synchronized(jobInitQueue){
  
• 
  if(jobInitQueue.size()>0){
  
• job=(JobInProgress)jobInitQueue.elementAt(0);
  
• jobInitQueue.remove(job);
  
• }else{
  
• 
  try{
  
• jobInitQueue.wait(JOBINIT_SLEEP_INTERVAL);
  
• }catch(InterruptedExceptioniex){
  
• }
  
• }
  
• }
  
• 
  try{
  
• 
  if(job!=null){
  
• job.initTasks();
  
• }
  
• }catch(Exceptione){
  
• LOG.log(Level.WARNING,"jobinitfailed",e);
  
• job.kill();
  
• }
  

synchronized (jobInitQueue) {
if (jobInitQueue.size() > 0) {
job = (JobInProgress) jobInitQueue.elementAt(0);
jobInitQueue.remove(job);
} else {
try {
jobInitQueue.wait(JOBINIT_SLEEP_INTERVAL);
} catch (InterruptedException iex) {
}
}
}
try {
if (job != null) {
job.initTasks();
}
} catch (Exception e) {
LOG.log(Level.WARNING, "job init failed", e);
job.kill();
}
　　

2、实现了两组rpc服务（协议），其中InterTrackerProtocol如下：
　　1）TaskTracker间隔几秒钟发送的心跳服务；

[java]
view plaincopyprint?

• 
  intemitHeartbeat(TaskTrackerStatusstatus,booleaninitialContact);
  

int emitHeartbeat(TaskTrackerStatus status, boolean initialContact);
　　2）向JobTracker获取新的任务；

[java]
view plaincopyprint?

• TaskpollForNewTask(StringtrackerName);
  

Task pollForNewTask(String trackerName);
　　3）询问JobTracker，任务是否可以终结；

[java]
view plaincopyprint?

• StringpollForTaskWithClosedJob(StringtrackerName);
  

String pollForTaskWithClosedJob(String trackerName);
　　4）Reduce Task询问JobTracker，哪些Map Task已经结束；

[java]
view plaincopyprint?

• MapOutputLocation[]locateMapOutputs(StringtaskId,String[][]mapTasksNeeded);
  

  MapOutputLocation[] locateMapOutputs(String taskId, String[][] mapTasksNeeded);

　　5）获取文件系统名；

[java]
view plaincopyprint?

• 
  publicStringgetFilesystemName()throwsIOException;
  

public String getFilesystemName() throws IOException;
　　JobSubmissionProtocol如下：
　　1）提交一个待执行的job；

[java]
view plaincopyprint?

• 
  publicJobStatussubmitJob(StringjobFile)throwsIOException;
  

public JobStatus submitJob(String jobFile) throws IOException;
　　2）杀死一个job；

[java]
view plaincopyprint?

• 
  publicvoidkillJob(Stringjobid);
  

public void killJob(String jobid);
　　3）获取job的名字、id等信息；

[java]
view plaincopyprint?

• 
  publicJobProfilegetJobProfile(Stringjobid);
  

public JobProfile getJobProfile(String jobid);
　　4）获取job的状态；

[java]
view plaincopyprint?

• 
  publicJobStatusgetJobStatus(Stringjobid);
  

public JobStatus getJobStatus(String jobid);
　　5）获取Map任务的报告；

[java]
view plaincopyprint?

• 
  publicTaskReport[]getMapTaskReports(Stringjobid);
  

public TaskReport[] getMapTaskReports(String jobid);
　　6）获取Reduce任务的报告；

[java]
view plaincopyprint?

• 
  publicTaskReport[]getReduceTaskReports(Stringjobid);