時間輪在Kafka的實踐

//在Systemtimer中添加一個任務，任務被包裝為一個TimerTaskEntryprivate def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {//先判斷是否可以添加進時間輪中，如果不可以添加進去代表任務已經過期或者任務被取消，注意這里的timingWheel持有上一層時間輪的引用，所以可能存在遞歸調用  if (!timingWheel.add(timerTaskEntry)) {    // Already expired or cancelled    if (!timerTaskEntry.cancelled)     //過期任務直接線程池異步執(zhí)行掉      taskExecutor.submit(timerTaskEntry.timerTask)  }}timingWheel添加任務，遞歸添加直到添加該任務進合適的時間輪的bucket中def add(timerTaskEntry: TimerTaskEntry): Boolean = {  val expiration = timerTaskEntry.expirationMs  //任務取消  if (timerTaskEntry.cancelled) {    // Cancelled    false  } else if (expiration < currentTime + tickMs) {    // 任務過期后會被執(zhí)行    false  } else if (expiration < currentTime + interval) {//任務過期時間比當前時間輪時間加周期小說明任務過期時間在本時間輪周期內    val virtualId = expiration / tickMs    //找到任務對應本時間輪的bucket    val bucket = buckets((virtualId % wheelSize.toLong).toInt)    bucket.add(timerTaskEntry)    // Set the bucket expiration time   //只有本bucket內的任務都過期后才會bucket.setExpiration返回true此時將bucket放入延遲隊列    if (bucket.setExpiration(virtualId * tickMs)) {     //bucket是一個TimerTaskList，它實現(xiàn)了java.util.concurrent.Delayed接口，里面是一個多任務組成的鏈表，圖2有說明      queue.offer(bucket)    }    true  } else {    // Out of the interval. Put it into the parent timer    //任務的過期時間不在本時間輪周期內說明需要升級時間輪，如果不存在則構造上一層時間輪，繼續(xù)用上一層時間輪添加任務    if (overflowWheel == null) addOverflowWheel()    overflowWheel.add(timerTaskEntry)  }}

private[this] def addOverflowWheel(): Unit = {  synchronized {    if (overflowWheel == null) {      overflowWheel = new TimingWheel(        tickMs = interval,        wheelSize = wheelSize,        startMs = currentTime,        taskCounter = taskCounter,        queue      )    }  }}

def advanceClock(timeMs: Long): Unit = {  if (timeMs >= currentTime + tickMs) {   //把當前時間打平為時間輪tickMs的整數(shù)倍    currentTime = timeMs - (timeMs % tickMs)    // Try to advance the clock of the overflow wheel if present    //驅動上層時間輪，這里的傳給上層的currentTime時間是本層時間輪打平過的，但是在上層時間輪還是會繼續(xù)打平    if (overflowWheel != null) overflowWheel.advanceClock(currentTime)  }}

//循環(huán)bucket里面的任務列表，一個個重新添加進時間輪，對符合條件的時間輪進行升降級或者執(zhí)行任務private[this] val reinsert = (timerTaskEntry: TimerTaskEntry) => addTimerTaskEntry(timerTaskEntry) /* * Advances the clock if there is an expired bucket. If there isn't any expired bucket when called, * waits up to timeoutMs before giving up. */def advanceClock(timeoutMs: Long): Boolean = {  var bucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS)  if (bucket != null) {    writeLock.lock()    try {      while (bucket != null) {        //驅動時間輪        timingWheel.advanceClock(bucket.getExpiration())       //循環(huán)buckek也就是任務列表，任務列表一個個繼續(xù)添加進時間輪以此來升級或者降級時間輪，把過期任務找出來執(zhí)行        bucket.flush(reinsert)       //循環(huán)        //這里就是從延遲隊列取出bucket，bucket是有延遲時間的，取出代表該bucket過期，我們通過bucket能取到bucket包含的任務列表        bucket = delayQueue.poll()      }    } finally {      writeLock.unlock()    }    true  } else {    false  }}