hadoop源码分析之MapReduce(二)

lenf

任务的申请、派发与执行
TaskTracker.run()连接JobTracker
　　TaskTracker的启动过程会初始化一系列参数和服务（另有单独的一节介绍），然后尝试连接JobTracker服务（即必须实现InterTrackerProtocol接口），如果连接断开，则会循环尝试连接JobTracker，并重新初始化所有成员和参数，此过程参见run()方法。
TaskTracker.offerService()主循环
　　如果连接JobTracker服务成功，TaskTracker就会调用offerService()函数进入主执行循环中。这个循环会每隔10秒与JobTracker通讯一次，调用transmitHeartBeat()获得HeartbeatResponse信息。然后调用HeartbeatResponse的getActions()函数获得JobTracker传过来的所有指令即一个TaskTrackerAction数组。再遍历这个数组，如果是一个新任务指令即LaunchTaskAction则调用startNewTask()函数执行新任务，
　　如果是CommitTaskAction
　　否则加入到tasksToCleanup队列，交给一个taskCleanupThread线程来处理，如执行KillJobAction或者KillTaskAction等。
TaskTracker.transmitHeartBeat()获取JobTracker指令
　　在transmitHeartBeat()函数处理中，TaskTracker会创建一个新的TaskTrackerStatus对象记录目前任务的执行状况，然后通过IPC接口调用JobTracker的heartbeat()方法发送过去，并接受新的指令，即返回值TaskTrackerAction数组。在这个调用之前，TaskTracker会先检查目前执行的Task数目以及本地磁盘的空间使用情况等，如果可以接收新的Task则设置heartbeat()的askForNewTask参数为true。操作成功后再更新相关的统计信息等。
JobTracker调度作业第二步：派发任务
JobTracker接到TaskTracker的heartbeat()调用后，首先会检查上一个心跳响应是否完成，是没要求启动或重启任务，如果一切正常，则会处理心跳。JobTracker会使用它的调度器taskScheduler来组装任务到一个任务列表tasks中。具体实现在taskScheduler的assignTasks()方法。得到tasks的数据后，把这些任务封装在一些LanuchTaskAction中，发回给TaskTracker，让它去执行任务。此时JobTracker的hearbeat()结束派发任务。
下面简单分析下hadoop默认的作业调度器JobQueueTaskScheduler怎么实现以上所说的assignTasks()方法。首先它会检查TaskTracker端还可以做多少个map和reduce任务，将要派发的任务数是否超出这个数，是否超出集群的任务平均剩余可负载数。如果都没超出，则为此TaskTracker分配一个MapTask或ReduceTask。产生Map任务使用JobInProgress的obtainNewMapTask()方法，实质上最后调用了JobInProgress的findNewMapTask()访问nonRunningMapCache得到Map任务的TaskInProgress；而产生Reduce任务使用JobInProgress.obtainNewReduceTask()方法，实质上最后调用了JobInProgress的findNewReduceTask()访问nonRuningReduceCache得到Reduce任务的TaskInProgress。