spark 源码分析之三 -- LiveListenerBus介绍

LiveListenerBus 

官方说明以下：

Asynchronously passes SparkListenerEvents to registered SparkListeners.

即它的功能是异步地将SparkListenerEvent传递给已经注册的SparkListener，这种异步的机制是经过生产消费者模型来实现的。

首先，它定义了 4 个 消息堵塞队列，队列的名字分别为shared、appStatus、executorManagement、eventLog。队列的类型是 org.apache.spark.scheduler.AsyncEventQueue#AsyncEventQueue，保存在 queues 变量中。每个队列上均可以注册监听器，若是队列没有监听器，则会被移除。

它有启动和stop和start两个标志位来指示 监听总线的的启动中止状态。 若是总线没有启动，有事件过来，先放到 一个待添加的可变数组中，不然直接将事件 post 到每个队列中。

其直接依赖类是 AsyncEventQueue， 至关于 LiveListenerBus 的多事件队列是对 AsyncEventQueue 进一步的封装。

AsyncEventQueue

其继承关系以下：

　　

它有启动和stop和start两个标志位来指示 监听总线的的启动中止状态。

其内部维护了listenersPlusTimers 主要就是用来保存注册到这个总线上的监听器对象的。

post 操做将事件放入内部的 LinkedBlockingQueue中，默认大小是 10000。

有一个事件分发器，它不停地从 LinkedBlockingQueue 执行 take 操做，获取事件，并将事件进一步分发给全部的监听器，由org.apache.spark.scheduler.SparkListenerBus#doPostEvent 方法实现事件转发，具体代码以下：

 1 protected override def doPostEvent(
 2  listener: SparkListenerInterface, 3 event: SparkListenerEvent): Unit = { 4  event match { 5 case stageSubmitted: SparkListenerStageSubmitted => 6  listener.onStageSubmitted(stageSubmitted) 7 case stageCompleted: SparkListenerStageCompleted => 8  listener.onStageCompleted(stageCompleted) 9 case jobStart: SparkListenerJobStart => 10  listener.onJobStart(jobStart) 11 case jobEnd: SparkListenerJobEnd => 12  listener.onJobEnd(jobEnd) 13 case taskStart: SparkListenerTaskStart => 14  listener.onTaskStart(taskStart) 15 case taskGettingResult: SparkListenerTaskGettingResult => 16  listener.onTaskGettingResult(taskGettingResult) 17 case taskEnd: SparkListenerTaskEnd => 18  listener.onTaskEnd(taskEnd) 19 case environmentUpdate: SparkListenerEnvironmentUpdate => 20  listener.onEnvironmentUpdate(environmentUpdate) 21 case blockManagerAdded: SparkListenerBlockManagerAdded => 22  listener.onBlockManagerAdded(blockManagerAdded) 23 case blockManagerRemoved: SparkListenerBlockManagerRemoved => 24  listener.onBlockManagerRemoved(blockManagerRemoved) 25 case unpersistRDD: SparkListenerUnpersistRDD => 26  listener.onUnpersistRDD(unpersistRDD) 27 case applicationStart: SparkListenerApplicationStart => 28  listener.onApplicationStart(applicationStart) 29 case applicationEnd: SparkListenerApplicationEnd => 30  listener.onApplicationEnd(applicationEnd) 31 case metricsUpdate: SparkListenerExecutorMetricsUpdate => 32  listener.onExecutorMetricsUpdate(metricsUpdate) 33 case executorAdded: SparkListenerExecutorAdded => 34  listener.onExecutorAdded(executorAdded) 35 case executorRemoved: SparkListenerExecutorRemoved => 36  listener.onExecutorRemoved(executorRemoved) 37 case executorBlacklistedForStage: SparkListenerExecutorBlacklistedForStage => 38  listener.onExecutorBlacklistedForStage(executorBlacklistedForStage) 39 case nodeBlacklistedForStage: SparkListenerNodeBlacklistedForStage => 40  listener.onNodeBlacklistedForStage(nodeBlacklistedForStage) 41 case executorBlacklisted: SparkListenerExecutorBlacklisted => 42  listener.onExecutorBlacklisted(executorBlacklisted) 43 case executorUnblacklisted: SparkListenerExecutorUnblacklisted => 44  listener.onExecutorUnblacklisted(executorUnblacklisted) 45 case nodeBlacklisted: SparkListenerNodeBlacklisted => 46  listener.onNodeBlacklisted(nodeBlacklisted) 47 case nodeUnblacklisted: SparkListenerNodeUnblacklisted => 48  listener.onNodeUnblacklisted(nodeUnblacklisted) 49 case blockUpdated: SparkListenerBlockUpdated => 50  listener.onBlockUpdated(blockUpdated) 51 case speculativeTaskSubmitted: SparkListenerSpeculativeTaskSubmitted => 52  listener.onSpeculativeTaskSubmitted(speculativeTaskSubmitted) 53 case _ => listener.onOtherEvent(event) 54  } 55 }

而后去调用 listener 的相对应的方法。

就这样，事件总线上的消息事件被监听器消费了。