trait SparkListenerEvent {
  /* Whether output this event to the event log */
  protected[spark] def logEvent: Boolean = true
}

abstract class SparkListener extends SparkListenerInterface {
  override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = { }
  override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted): Unit = { }
  override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = { }
  override def onTaskGettingResult(taskGettingResult: SparkListenerTaskGettingResult): Unit = { }
  override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { }
  override def onJobStart(jobStart: SparkListenerJobStart): Unit = { }
  override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = { }
  override def onEnvironmentUpdate(environmentUpdate: SparkListenerEnvironmentUpdate): Unit = { }
  override def onBlockManagerAdded(blockManagerAdded: SparkListenerBlockManagerAdded): Unit = { }
  override def onBlockManagerRemoved(
      blockManagerRemoved: SparkListenerBlockManagerRemoved): Unit = { }
  override def onUnpersistRDD(unpersistRDD: SparkListenerUnpersistRDD): Unit = { }
  override def onApplicationStart(applicationStart: SparkListenerApplicationStart): Unit = { }
  override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = { }
  override def onExecutorMetricsUpdate(
      executorMetricsUpdate: SparkListenerExecutorMetricsUpdate): Unit = { }
  override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded): Unit = { }
  override def onExecutorRemoved(executorRemoved: SparkListenerExecutorRemoved): Unit = { }
  override def onBlockUpdated(blockUpdated: SparkListenerBlockUpdated): Unit = { }
  override def onOtherEvent(event: SparkListenerEvent): Unit = { }
}

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

  protected override def doPostEvent(
      listener: SparkListenerInterface,
      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 blockUpdated: SparkListenerBlockUpdated =>
        listener.onBlockUpdated(blockUpdated)
      case logStart: SparkListenerLogStart => // ignore event log metadata
      case _ => listener.onOtherEvent(event)
    }
  }

}

private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging {

  // Marked `private[spark]` for access in tests.
  private[spark] val listeners = new CopyOnWriteArrayList[L]

  final def addListener(listener: L): Unit = {
    listeners.add(listener)
  }

  final def removeListener(listener: L): Unit = {
    listeners.remove(listener)
  }

  /**
   * Post the event to all registered listeners. The `postToAll` caller should guarantee calling
   * `postToAll` in the same thread for all events.
   */
  final def postToAll(event: E): Unit = {
    // JavaConverters can create a JIterableWrapper if we use asScala.
    // However, this method will be called frequently. To avoid the wrapper cost, here we use
    // Java Iterator directly.
    val iter = listeners.iterator
    while (iter.hasNext) {
      val listener = iter.next()
      try {
        doPostEvent(listener, event)
      } catch {
        case NonFatal(e) =>
          logError(s"Listener ${Utils.getFormattedClassName(listener)} threw an exception", e)
      }
    }
  }

  /**
   * Post an event to the specified listener. `onPostEvent` is guaranteed to be called in the same
   * thread for all listeners.
   */
  protected def doPostEvent(listener: L, event: E): Unit

  private[spark] def findListenersByClass[T <: L : ClassTag](): Seq[T] = {
    val c = implicitly[ClassTag[T]].runtimeClass
    listeners.asScala.filter(_.getClass == c).map(_.asInstanceOf[T]).toSeq
  }

}

// Cap the capacity of the event queue so we get an explicit error (rather than
// an OOM exception) if it's perpetually being added to more quickly than it's being drained.
private lazy val EVENT_QUEUE_CAPACITY = validateAndGetQueueSize()
private lazy val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)

private val listenerThread = new Thread(name) {  // <-- 线程名为SparkListenerBus
  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
          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
          }
        }
      }
    }
  }
}

def post(event: SparkListenerEvent): Unit = {
  if (stopped.get) {
    // Drop further events to make `listenerThread` exit ASAP
    logError(s"$name has already stopped! Dropping event $event")
    return
  }
  val eventAdded = eventQueue.offer(event)  // <-- 这里将新来的事件添加到消息队列中
  if (eventAdded) {
    eventLock.release()
  } else {
    onDropEvent(event)  // <-- 没有添加成功，则丢弃事件
    droppedEventsCounter.incrementAndGet()
  }

  val droppedEvents = droppedEventsCounter.get
  if (droppedEvents > 0) {
    // Don't log too frequently
    if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {
      // There may be multiple threads trying to decrease droppedEventsCounter.
      // Use "compareAndSet" to make sure only one thread can win.
      // And if another thread is increasing droppedEventsCounter, "compareAndSet" will fail and
      // then that thread will update it.
      if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {
        val prevLastReportTimestamp = lastReportTimestamp
        lastReportTimestamp = System.currentTimeMillis()
        logWarning(s"Dropped $droppedEvents SparkListenerEvents since " +
          new java.util.Date(prevLastReportTimestamp))
      }
    }
  }
}

// An asynchronous listener bus for Spark events
private[spark] val listenerBus = new LiveListenerBus(this)
随后创建各种listener，并注册到listenerBus中，通过调用listenerBus的start()方法启动消息分发流程。
private def setupAndStartListenerBus(): Unit = {
  // Use reflection to instantiate listeners specified via `spark.extraListeners`
  try {
    val listenerClassNames: Seq[String] =
      conf.get("spark.extraListeners", "").split(',').map(_.trim).filter(_ != "")
    for (className <- listenerClassNames) {  // <-- 如果指定了额外的SparkListenr类，可通过反射机制实例化并注册到listenerBus
      // Use reflection to find the right constructor
      val constructors = {
        val listenerClass = Utils.classForName(className)
        listenerClass
            .getConstructors
            .asInstanceOf[Array[Constructor[_ <: SparkListenerInterface]]]
      }
      val constructorTakingSparkConf = constructors.find { c =>
        c.getParameterTypes.sameElements(Array(classOf[SparkConf]))
      }
      lazy val zeroArgumentConstructor = constructors.find { c =>
        c.getParameterTypes.isEmpty
      }
      val listener: SparkListenerInterface = {
        if (constructorTakingSparkConf.isDefined) {
          constructorTakingSparkConf.get.newInstance(conf)
        } else if (zeroArgumentConstructor.isDefined) {
          zeroArgumentConstructor.get.newInstance()
        } else {
          ...
        }
      }
      listenerBus.addListener(listener)
      logInfo(s"Registered listener $className")
    }
  } catch {
    ...
  }

  listenerBus.start()
  _listenerBusStarted = true
}

def start(): Unit = {
  if (started.compareAndSet(false, true)) {
    listenerThread.start()
  } else {
    throw new IllegalStateException(s"$name already started!")
  }
}

def stop(): Unit = {
  if (!started.get()) {
    throw new IllegalStateException(s"Attempted to stop $name that has not yet started!")
  }
  if (stopped.compareAndSet(false, true)) {
    // Call eventLock.release() so that listenerThread will poll `null` from `eventQueue` and know
    // `stop` is called.
    eventLock.release()
    listenerThread.join()
  } else {
    // Keep quiet
  }
}