Netty-時間輪

歸檔

• GitHub: Netty-時間輪

---

參考

• Netty-時間輪

---

說明

• 其實 Netty 框架并沒有使用，其可做學習算法原理的參考

---

單元測試

public class HashedWheelTimerTest2 {public static void main(String[] args) {System.out.println("---------> " + LocalTime.now());Timer timer = new HashedWheelTimer(); // sign_o_001Timeout timeout1 = timer.newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) {System.out.println("timeout1: " + LocalTime.now());}}, 10, TimeUnit.SECONDS);if (!timeout1.isExpired()) {timeout1.cancel(); // 取消任務}timer.newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws InterruptedException {System.out.println("timeout2: " + LocalTime.now());Thread.sleep(5000); // 占用 5 s，后面的任務會被延后}}, 1, TimeUnit.SECONDS);timer.newTimeout(new TimerTask() { // sign_c_001@Overridepublic void run(Timeout timeout) {System.out.println("timeout3: " + LocalTime.now());System.out.println(timeout.timer()); // 即：sign_o_001System.out.println(timeout.task().getClass()); // 即：sign_c_001}}, 3, TimeUnit.SECONDS);System.out.println("timer -> " + timer);System.out.println("---------> " + LocalTime.now());}
}

---

原理

類結構

• io.netty.util.HashedWheelTimer

/*** 時間輪 */
public class HashedWheelTimer implements Timer {private final Worker worker = new Worker(); // 任務執行體private final Thread workerThread;          // 任務線程private final long tickDuration;            // 時針每次 tick 的時間，相當于時針間隔多久走到下一個 slotprivate final HashedWheelBucket[] wheel;    // 槽位數組private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue(); // MPSC 隊列，方便添加處理（用于削峰） sign_f_001private final Queue<HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue(); // 記錄取消的，方便移除private final AtomicLong pendingTimeouts = new AtomicLong(0); // 記錄等待的任務數/** 槽位 */private static final class HashedWheelBucket {private HashedWheelTimeout head;    // 頭private HashedWheelTimeout tail;    // 尾}/** 任務封裝 */private static final class HashedWheelTimeout implements Timeout, Runnable {private final TimerTask task;   // 要執行的任務體private final long deadline;    // 終止時間戳long remainingRounds;           // 剩余輪數HashedWheelTimeout next;        // 與 prev 組成雙向鏈HashedWheelTimeout prev;}   
}

調用鏈

• io.netty.util.HashedWheelTimer

/*** 時間輪 */
public class HashedWheelTimer implements Timer {/** 添加任務 */@Overridepublic Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {... // 省略校驗start(); // 啟動線程（如果還沒啟動的話）long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; // 計算終止時間(startTime 一般為 1)... // 省略 deadline 溢值處理HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);timeouts.add(timeout); // 添加到削峰隊列 sign_f_001return timeout;}
}

• io.netty.util.HashedWheelTimer.Worker

    /** 任務執行體 */private final class Worker implements Runnable {private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>(); // 方便 stop() 后記錄并返回private long tick; // 走時計數@Overridepublic void run() {... // 省略其他處理do {final long deadline = waitForNextTick();    // 計算并等待if (deadline > 0) {int idx = (int) (tick & mask);  // 計算要處理的槽位索引processCancelledTasks();        // 清空被取消的任務HashedWheelBucket bucket = wheel[idx];transferTimeoutsToBuckets();    // 將隊列中的任務轉移到相應的槽位里bucket.expireTimeouts(deadline);// 執行槽位里面的任務 sign_m_101tick++; // 計數加 1}} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED); // ... // 省略 stop 處理}// 計算并等待private long waitForNextTick() {long deadline = tickDuration * (tick + 1);for (;;) {final long currentTime = System.nanoTime() - startTime;long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;if (sleepTimeMs <= 0) { // 小于 0 表示不用等待if (currentTime == Long.MIN_VALUE) {return -Long.MAX_VALUE; // 返回最小值，表示后面的邏輯：任務都執行} else {return currentTime;     // 返回當前時間用于參考}}... // 省略 Windows 平臺處理try {Thread.sleep(sleepTimeMs);  // 線程睡眠} ... // catch 中斷異常處理}}// 轉移任務private void transferTimeoutsToBuckets() {for (int i = 0; i < 10_0000; i++) {HashedWheelTimeout timeout = timeouts.poll();... // 省略無新增的任務或任務被取消的處理long calculated = timeout.deadline / tickDuration;timeout.remainingRounds = (calculated - tick) / wheel.length; // 第幾輪數final long ticks = Math.max(calculated, tick);int stopIndex = (int) (ticks & mask); // 槽位索引HashedWheelBucket bucket = wheel[stopIndex];bucket.addTimeout(timeout);}}}

• io.netty.util.HashedWheelTimer.HashedWheelBucket

    /** 槽位 */private static final class HashedWheelBucket {/** 執行任務 sign_m_101 */public void expireTimeouts(long deadline) {HashedWheelTimeout timeout = head;while (timeout != null) {HashedWheelTimeout next = timeout.next;if (timeout.remainingRounds <= 0) {     // 輪數為 0 或小于 0，才執行任務next = remove(timeout);if (timeout.deadline <= deadline) { // 小于當前時間（正常流程）timeout.expire();               // 執行任務 sign_m_201} ... // else 處理（狀態不對，拋異常），一般不會發生} else if (timeout.isCancelled()) {next = remove(timeout);} else {timeout.remainingRounds--;          // 減任務輪數}timeout = next;}}}

• io.netty.util.HashedWheelTimer.HashedWheelTimeout

    /** 任務封裝 */private static final class HashedWheelTimeout implements Timeout, Runnable {/** 執行任務 sign_m_201 */public void expire() {if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {return; // 可能已取消}try {timer.taskExecutor.execute(this); // 執行具體的任務} catch (Throwable t) {... // 任務執行異常，只作日志記錄(warn 級別)}}}