如何進(jìn)行DAGScheduler源碼解讀，針對(duì)這個(gè)問(wèn)題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。

創(chuàng)新互聯(lián)建站專注于鎮(zhèn)原網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。 熱誠(chéng)為您提供鎮(zhèn)原營(yíng)銷型網(wǎng)站建設(shè)，鎮(zhèn)原網(wǎng)站制作、鎮(zhèn)原網(wǎng)頁(yè)設(shè)計(jì)、鎮(zhèn)原網(wǎng)站官網(wǎng)定制、小程序制作服務(wù)，打造鎮(zhèn)原網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供鎮(zhèn)原網(wǎng)站排名全網(wǎng)營(yíng)銷落地服務(wù)。

當(dāng)構(gòu)建完TaskScheduler之后，我們需要構(gòu)建DAGScheduler這個(gè)核心對(duì)象：

進(jìn)入其構(gòu)造函數(shù)中：

可以看出構(gòu)建DAGScheduler實(shí)例的時(shí)候需要把TaskScheduler實(shí)例對(duì)象作為參數(shù)傳入。

LiveListenerBus：

BlockManagerMaster：

通過(guò)閱讀代碼，我們可以發(fā)現(xiàn)DAGScheduler實(shí)例化的時(shí)候，調(diào)用了initializeEventProcessActor()方法

private def initializeEventProcessActor() {  // blocking the thread until supervisor is started, which ensures eventProcessActor is  // not null before any job is submitted  // 阻塞當(dāng)前線程，等待supervisor啟動(dòng)，這樣可以確保Job提交時(shí)，eventProcessActor not null  implicit val timeout = Timeout(30 seconds)  val initEventActorReply =    dagSchedulerActorSupervisor ? Props(new DAGSchedulerEventProcessActor(this))  eventProcessActor = Await.result(initEventActorReply, timeout.duration).
    asInstanceOf[ActorRef]
}

initializeEventProcessActor()

DAGSchedulerEventProcessActor：

private[scheduler] class DAGSchedulerEventProcessActor(dagScheduler: DAGScheduler)  extends Actor with Logging {  override def preStart() {    // set DAGScheduler for taskScheduler to ensure eventProcessActor is always    // valid when the messages arrive    // 設(shè)置taskScheduler對(duì)DAGScheduler的引用句柄。在此處設(shè)置保證了Job提交時(shí)候    // eventProcessActor已經(jīng)準(zhǔn)備就緒    dagScheduler.taskScheduler.setDAGScheduler(dagScheduler)
  }  /**   * The main event loop of the DAG scheduler.   */  def receive = {    case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties) =>
      dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite,        listener, properties)    case StageCancelled(stageId) =>
      dagScheduler.handleStageCancellation(stageId)    case JobCancelled(jobId) =>
      dagScheduler.handleJobCancellation(jobId)    case JobGroupCancelled(groupId) =>
      dagScheduler.handleJobGroupCancelled(groupId)    case AllJobsCancelled =>
      dagScheduler.doCancelAllJobs()    case ExecutorAdded(execId, host) =>
      dagScheduler.handleExecutorAdded(execId, host)    case ExecutorLost(execId) =>
      dagScheduler.handleExecutorLost(execId, fetchFailed = false)    case BeginEvent(task, taskInfo) =>
      dagScheduler.handleBeginEvent(task, taskInfo)    case GettingResultEvent(taskInfo) =>
      dagScheduler.handleGetTaskResult(taskInfo)    case completion @ CompletionEvent(task, reason, _, _, taskInfo, taskMetrics) =>
      dagScheduler.handleTaskCompletion(completion)    case TaskSetFailed(taskSet, reason) =>
      dagScheduler.handleTaskSetFailed(taskSet, reason)    case ResubmitFailedStages =>
      dagScheduler.resubmitFailedStages()
  }  override def postStop() {    // Cancel any active jobs in postStop hook    dagScheduler.cleanUpAfterSchedulerStop()
  }
}

可以看出核心在于實(shí)例化eventProcessActor對(duì)象，eventProcessActor會(huì)負(fù)責(zé)接收和發(fā)送DAGScheduler的消息，是DAGScheduler的通信載體。

關(guān)于如何進(jìn)行DAGScheduler源碼解讀問(wèn)題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒(méi)有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識(shí)。

新聞標(biāo)題：如何進(jìn)行DAGScheduler源碼解讀
文章路徑：http://jinyejixie.com/article0/gdjsio.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站設(shè)計(jì)公司、靜態(tài)網(wǎng)站、網(wǎng)頁(yè)設(shè)計(jì)公司、面包屑導(dǎo)航、App設(shè)計(jì)、虛擬主機(jī)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)