Spark2.1.0事件匯流排分析——LiveListenerBus詳解

LiveListenerBus繼承了SparkListenerBus，並實現了將事件非同步投遞給監聽器，達到實時刷新UI界面數據的效果。LiveListenerBus主要由以下部分組成：

• eventQueue：是SparkListenerEvent事件的阻塞隊列，隊列大小可以通過Spark屬性spark.scheduler.listenerbus.eventqueue.size進行配置，默認為10000（Spark早期版本中屬於靜態屬性，固定為10000，這導致隊列堆滿時，只得移除一些最老的事件，最終導致各種問題與bug）；
• started：標記LiveListenerBus的啟動狀態的AtomicBoolean類型的變數；
• stopped：標記LiveListenerBus的停止狀態的AtomicBoolean類型的變數；
• droppedEventsCounter：使用AtomicLong類型對刪除的事件進行計數，每當日誌列印了droppedEventsCounter後，會將droppedEventsCounter重置為0；
• lastReportTimestamp：用於記錄最後一次日誌列印droppedEventsCounter的時間戳；
• processingEvent：用來標記當前正有事件被listenerThread線程處理；
• logDroppedEvent：AtomicBoolean類型的變數，用於標記是否由於eventQueue已滿，導致新的事件被刪除；
• eventLock：用於當有新的事件到來時釋放信號量，當對事件進行處理時獲取信號量；
• listeners：繼承自LiveListenerBus的監聽器數組；
• listenerThread：處理事件的線程。

非同步事件處理線程

listenerThread用於非同步處理eventQueue中的事件，為了便於說明，這裡將展示listenerThread及LiveListenerBus中的主要代碼片段，見代碼清單1。

代碼清單1 LiveListenerBus主要邏輯的代碼片段

private lazy val EVENT_QUEUE_CAPACITY = validateAndGetQueueSize()
private lazy val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)

private def validateAndGetQueueSize(): Int = {
val queueSize = sparkContext.conf.get(LISTENER_BUS_EVENT_QUEUE_SIZE)
if (queueSize <= 0) {
throw new SparkException("spark.scheduler.listenerbus.eventqueue.size must be > 0!")
}
queueSize
}

private val started = new AtomicBoolean(false)
private val stopped = new AtomicBoolean(false)
private val droppedEventsCounter = new AtomicLong(0L)
@volatile private var lastReportTimestamp = 0L
private var processingEvent = false
private val logDroppedEvent = new AtomicBoolean(false)
private val eventLock = new Semaphore(0)

private val listenerThread = new Thread(name) {
setDaemon(true)
override def run(): Unit = Utils.tryOrStopSparkContext(sparkContext) {
LiveListenerBus.withinListenerThread.withValue(true) {
while (true) {
eventLock.acquire() // 獲取信號量
self.synchronized {
processingEvent = true
}
try {
val event = eventQueue.poll //從eventQueue中獲取事件
if (event == null) {
// Get out of the while loop and shutdown the daemon thread
if (!stopped.get) {
throw new IllegalStateException("Polling `null` from eventQueue means" +
" the listener bus has been stopped. So `stopped` must be true")
}
return
}
postToAll(event) // 事件處理
} finally {
self.synchronized {
processingEvent = false
}
}
}
}
}
}

通過分析代碼清單1，listenerThread的工作步驟為：

1. 不斷獲取信號量（當可以獲取信號量時，說明還有事件未處理）；
2. 通過同步控制，將processingEvent設置為true；
3. 從eventQueue中獲取事件；
4. 調用超類ListenerBus的postToAll方法（postToAll方法對監聽器進行遍歷，並調用SparkListenerBus的doPostEvent方法對事件進行匹配後執行監聽器的相應方法）；
5. 每次循環結束依然需要通過同步控制，將processingEvent設置為false；

值得一提的是，listenerThread的run方法中調用了Utils的tryOrStopSparkContext，tryOrStopSparkContext方法可以保證當listenerThread的內部循環拋出異常後啟動一個新的線程停止SparkContext（SparkContext的內容將在第4章詳細介紹，tryOrStopSparkContext方法的具體實現請閱讀《附錄A Spark2.1核心工具類Utils》）。

LiveListenerBus的消息投遞

在解釋了非同步線程listenerThread的工作內容後，還有一個要點沒有解釋：eventQueue中的事件是如何放進去的呢？由於eventQueue定義在LiveListenerBus中，因此ListenerBus和SparkListenerBus中並沒有操縱eventQueue的方法，要將事件放入eventQueue只能依靠LiveListenerBus自己了，其post方法就是為此目的而生的，見代碼清單2。

代碼清單2 向LiveListenerBus投遞SparkListenerEvent事件

def post(event: SparkListenerEvent): Unit = {
if (stopped.get) {
logError(s"$name has already stopped! Dropping event $event")
return
}
val eventAdded = eventQueue.offer(event) // 向eventQueue中添加事件
if (eventAdded) {
eventLock.release()
} else {
onDropEvent(event)
droppedEventsCounter.incrementAndGet()
}
// 列印刪除事件數的日誌
val droppedEvents = droppedEventsCounter.get
if (droppedEvents > 0) {
if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {
if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {
val prevLastReportTimestamp = lastReportTimestamp
lastReportTimestamp = System.currentTimeMillis()
logWarning(s"Dropped $droppedEvents SparkListenerEvents since " +
new java.util.Date(prevLastReportTimestamp))
}
}
}
}

從代碼清單2看到post方法的處理步驟如下：

1. 判斷LiveListenerBus是否已經處於停止狀態；
2. 向eventQueue中添加事件。如果添加成功，則釋放信號量進而催化listenerThread能夠有效工作。如果eventQueue已滿造成添加失敗，則移除事件，並對刪除事件計數器droppedEventsCounter進行自增；
3. 如果有事件被刪除，並且當前系統時間距離上一次列印droppedEventsCounter超過了60秒則將droppedEventsCounter列印到日誌。

LiveListenerBus與監聽器

與LiveListenerBus配合使用的監聽器，並非是父類SparkListenerBus的類型參數SparkListenerInterface，而是繼承自SparkListenerInterface的SparkListener及其子類。圖1列出了Spark中監聽器SparkListener以及它的6種最常用的實現[1]。

圖1 SparkListener的類繼承體系

SparkListener雖然實現了SparkListenerInterface中的每個方法，但是其實都是空實現，具體的實現需要交給子類去完成。

《Spark2.1.0之源碼分析——事件匯流排》中首先對事件匯流排的介面定義進行了一些介紹，之後《Spark2.1.0事件匯流排分析——ListenerBus的繼承體系》一文展示了ListenerBus的繼承體系，然後《Spark2.1.0事件匯流排分析——SparkListenerBus詳解》選擇ListenerBus的子類SparkListenerBus進行分析，最後本文選擇LiveListenerBus作為具體的實現例子進行分析，這裡將通過圖2更加直觀的展示ListenerBus、SparkListenerBus及LiveListenerBus的工作原理。

圖2 LiveListenerBus的工作流程圖

最後對於圖2作一些補充說明：圖中的DAGScheduler、SparkContext、BlockManagerMasterEndpoint、DriverEndpoint及LocalSchedulerBackend都是LiveListenerBus的事件來源，它們都是通過調用LiveListenerBus的post方法將消息交給非同步線程listenerThread處理的。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！