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 级别)}}}