死磕 Spark 事件总线——聊聊 Spark 中事件监听是如何实现的

  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  //获取事件， remove() 和 poll() 方法都是从队列中删除第一个元素（head）。            if (event == null) {              // 此时说明没有事件，但还是拿到信号量了，这说明stop方法被调用了              // 跳出while循环，关闭守护进程线程              if (!stopped.get) {                throw new IllegalStateException("Polling `null` from eventQueue means" +                  " the listener bus has been stopped. So `stopped` must be true")              }              return            }            // 调用ListenerBus的postxToAll(event: E)方法            postxToAll(event)          } finally {            self.synchronized {              processingEvent = false            }          }        }      }    }  }

private val started = new AtomicBoolean(false)private val stopped = new AtomicBoolean(false)//存放事件private lazy val eventQueue = new LinkedBlockingQueue[SparkListenerEvent]// 表示队列中产生和使用的事件数量的计数器，这个信号量是为了避免消费者线程空跑private val eventLock = new Semaphore(0)

  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.      // 释放一个信号量，但此时是没有事件的，从而listenerThread会拿到一个空事件，从而知道该停止了      eventLock.release()      //然后等待消费者线程自动关闭      listenerThread.join()    } else {      // Keep quiet    }  }

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    }    // 在事件队列队尾添加事件    // add（）和offer（）区别：两者都是往队列尾部插入元素，不同的时候，当超出队列界限的时候，add（）方法是抛出异常让你处理，而offer（）方法是直接返回false    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   日志不要太频繁      // 如果上一次，队列满了EVENT_QUEUE_CAPACITY=1000设置的值，就丢掉，然后记录一个时间，如果一直持续丢掉，那么每过60秒记录一次日志，不然日志会爆满的      if (System.currentTimeMillis() - lastReportTimestamp >= 60 * 1000) {        if (droppedEventsCounter.compareAndSet(droppedEvents, 0)) {          val prevLastReportTimestamp = lastReportTimestamp          lastReportTimestamp = System.currentTimeMillis()          // 记录一个warn日志，表示这个事件，被丢弃了          logWarning(s"Dropped $droppedEvents SparkListenerEvents since " +            new java.util.Date(prevLastReportTimestamp))        }      }    }  }