spark源码系列之内部通讯的三种机制

private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
 case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
    dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)

 case MapStageSubmitted(jobId, dependency, callSite, listener, properties) =>
    dagScheduler.handleMapStageSubmitted(jobId, dependency, 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, exception) =>
    dagScheduler.handleTaskSetFailed(taskSet, reason, exception)

 case ResubmitFailedStages =>
    dagScheduler.resubmitFailedStages()
}

final def postToAll(event: E): Unit = {

 val iter = listeners.iterator
 while (iter.hasNext) {
 val listener = iter.next()
 try {
 onPostEvent(listener, event)
 } catch {
 case NonFatal(e) =>
        logError(s"Listener ${Utils.getFormattedClassName(listener)} threw an exception", e)
    }
  }
}

private[spark] trait SparkListenerBus extends ListenerBus[SparkListener, SparkListenerEvent] {

 override def onPostEvent(listener: SparkListener, event: SparkListenerEvent): Unit = {
    event match {
 case stageSubmitted: SparkListenerStageSubmitted =>
        listener.onStageSubmitted(stageSubmitted)
 case stageCompleted: SparkListenerStageCompleted =>
        listener.onStageCompleted(stageCompleted)
 case jobStart: SparkListenerJobStart =>
        listener.onJobStart(jobStart)
 case jobEnd: SparkListenerJobEnd =>
        listener.onJobEnd(jobEnd)
 case taskStart: SparkListenerTaskStart =>
        listener.onTaskStart(taskStart)
 case taskGettingResult: SparkListenerTaskGettingResult =>
        listener.onTaskGettingResult(taskGettingResult)
 case taskEnd: SparkListenerTaskEnd =>
        listener.onTaskEnd(taskEnd)
 case environmentUpdate: SparkListenerEnvironmentUpdate =>
        listener.onEnvironmentUpdate(environmentUpdate)
 case blockManagerAdded: SparkListenerBlockManagerAdded =>
        listener.onBlockManagerAdded(blockManagerAdded)
 case blockManagerRemoved: SparkListenerBlockManagerRemoved =>
        listener.onBlockManagerRemoved(blockManagerRemoved)
 case unpersistRDD: SparkListenerUnpersistRDD =>
        listener.onUnpersistRDD(unpersistRDD)
 case applicationStart: SparkListenerApplicationStart =>
        listener.onApplicationStart(applicationStart)
 case applicationEnd: SparkListenerApplicationEnd =>
        listener.onApplicationEnd(applicationEnd)
 case metricsUpdate: SparkListenerExecutorMetricsUpdate =>
        listener.onExecutorMetricsUpdate(metricsUpdate)
 case executorAdded: SparkListenerExecutorAdded =>
        listener.onExecutorAdded(executorAdded)
 case executorRemoved: SparkListenerExecutorRemoved =>
        listener.onExecutorRemoved(executorRemoved)
 case blockUpdated: SparkListenerBlockUpdated =>
        listener.onBlockUpdated(blockUpdated)
 case logStart: SparkListenerLogStart => // ignore event log metadata
 }
  }

}

private[spark] class StreamingListenerBus
 extends AsynchronousListenerBus[StreamingListener, StreamingListenerEvent]("StreamingListenerBus")
 with Logging {

 private val logDroppedEvent = new AtomicBoolean(false)

 override def onPostEvent(listener: StreamingListener, event: StreamingListenerEvent): Unit = {
    event match {
 case receiverStarted: StreamingListenerReceiverStarted =>
        listener.onReceiverStarted(receiverStarted)
 case receiverError: StreamingListenerReceiverError =>
        listener.onReceiverError(receiverError)
 case receiverStopped: StreamingListenerReceiverStopped =>
        listener.onReceiverStopped(receiverStopped)
 case batchSubmitted: StreamingListenerBatchSubmitted =>
        listener.onBatchSubmitted(batchSubmitted)
 case batchStarted: StreamingListenerBatchStarted =>
        listener.onBatchStarted(batchStarted)
 case batchCompleted: StreamingListenerBatchCompleted =>
        listener.onBatchCompleted(batchCompleted)
 case outputOperationStarted: StreamingListenerOutputOperationStarted =>
        listener.onOutputOperationStarted(outputOperationStarted)
 case outputOperationCompleted: StreamingListenerOutputOperationCompleted =>
        listener.onOutputOperationCompleted(outputOperationCompleted)
 case _ =>
    }
  }

 override def onDropEvent(event: StreamingListenerEvent): Unit = {
 if (logDroppedEvent.compareAndSet(false, true)) {
 // Only log the following message once to avoid duplicated annoying logs.
 logError("Dropping StreamingListenerEvent because no remaining room in event queue. " +
 "This likely means one of the StreamingListeners is too slow and cannot keep up with the " +
 "rate at which events are being started by the scheduler.")
    }
  }
}