時(shí)間輪在Kafka的實(shí)踐

//在Systemtimer中添加一個(gè)任務(wù)，任務(wù)被包裝為一個(gè)TimerTaskEntryprivate def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {//先判斷是否可以添加進(jìn)時(shí)間輪中，如果不可以添加進(jìn)去代表任務(wù)已經(jīng)過(guò)期或者任務(wù)被取消，注意這里的timingWheel持有上一層時(shí)間輪的引用，所以可能存在遞歸調(diào)用  if (!timingWheel.add(timerTaskEntry)) {    // Already expired or cancelled    if (!timerTaskEntry.cancelled)     //過(guò)期任務(wù)直接線程池異步執(zhí)行掉      taskExecutor.submit(timerTaskEntry.timerTask)  }}timingWheel添加任務(wù)，遞歸添加直到添加該任務(wù)進(jìn)合適的時(shí)間輪的bucket中def add(timerTaskEntry: TimerTaskEntry): Boolean = {  val expiration = timerTaskEntry.expirationMs  //任務(wù)取消  if (timerTaskEntry.cancelled) {    // Cancelled    false  } else if (expiration < currentTime + tickMs) {    // 任務(wù)過(guò)期后會(huì)被執(zhí)行    false  } else if (expiration < currentTime + interval) {//任務(wù)過(guò)期時(shí)間比當(dāng)前時(shí)間輪時(shí)間加周期小說(shuō)明任務(wù)過(guò)期時(shí)間在本時(shí)間輪周期內(nèi)    val virtualId = expiration / tickMs    //找到任務(wù)對(duì)應(yīng)本時(shí)間輪的bucket    val bucket = buckets((virtualId % wheelSize.toLong).toInt)    bucket.add(timerTaskEntry)    // Set the bucket expiration time   //只有本bucket內(nèi)的任務(wù)都過(guò)期后才會(huì)bucket.setExpiration返回true此時(shí)將bucket放入延遲隊(duì)列    if (bucket.setExpiration(virtualId * tickMs)) {     //bucket是一個(gè)TimerTaskList，它實(shí)現(xiàn)了java.util.concurrent.Delayed接口，里面是一個(gè)多任務(wù)組成的鏈表，圖2有說(shuō)明      queue.offer(bucket)    }    true  } else {    // Out of the interval. Put it into the parent timer    //任務(wù)的過(guò)期時(shí)間不在本時(shí)間輪周期內(nèi)說(shuō)明需要升級(jí)時(shí)間輪，如果不存在則構(gòu)造上一層時(shí)間輪，繼續(xù)用上一層時(shí)間輪添加任務(wù)    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) {   //把當(dāng)前時(shí)間打平為時(shí)間輪tickMs的整數(shù)倍    currentTime = timeMs - (timeMs % tickMs)    // Try to advance the clock of the overflow wheel if present    //驅(qū)動(dòng)上層時(shí)間輪，這里的傳給上層的currentTime時(shí)間是本層時(shí)間輪打平過(guò)的，但是在上層時(shí)間輪還是會(huì)繼續(xù)打平    if (overflowWheel != null) overflowWheel.advanceClock(currentTime)  }}

//循環(huán)bucket里面的任務(wù)列表，一個(gè)個(gè)重新添加進(jìn)時(shí)間輪，對(duì)符合條件的時(shí)間輪進(jìn)行升降級(jí)或者執(zhí)行任務(wù)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) {        //驅(qū)動(dòng)時(shí)間輪        timingWheel.advanceClock(bucket.getExpiration())       //循環(huán)buckek也就是任務(wù)列表，任務(wù)列表一個(gè)個(gè)繼續(xù)添加進(jìn)時(shí)間輪以此來(lái)升級(jí)或者降級(jí)時(shí)間輪，把過(guò)期任務(wù)找出來(lái)執(zhí)行        bucket.flush(reinsert)       //循環(huán)        //這里就是從延遲隊(duì)列取出bucket，bucket是有延遲時(shí)間的，取出代表該bucket過(guò)期，我們通過(guò)bucket能取到bucket包含的任務(wù)列表        bucket = delayQueue.poll()      }    } finally {      writeLock.unlock()    }    true  } else {    false  }}