基于时间轮的定时器HashedWheelTimer

一、前言

最近在阅读Redisson的源码时看到了内部使用了netty提供的这个组件&＃xff0c;就想着看下这个定时器具体是如何实现的&＃xff1f;

先介绍一下HashedWheelTimer&＃xff0c;它是基于时间轮实现的一个定时器&＃xff0c;它的优点是实现相对简单&＃xff0c;缺点是无法精确、准时地执行定时任务&＃xff0c;只能是近似执行。

因为时间轮中每个刻度大小可能是100ms也可能1ms&＃xff0c;所以在执行任务时&＃xff0c;时间上会存在一点误差&＃xff0c;在大部分网络应用中&＃xff0c;IO任务的执行时间往往不需要那么精确&＃xff0c;因此默认每个刻度小大是100ms&＃xff0c;但你可以自己来调整刻度大小&＃xff0c;最小是1ms。

简单介绍完HahsedWheelTimer&＃xff0c;接下来我们先来看下时间轮的结构

二、时间轮的结构

时间轮类似于一个时钟&＃xff0c;它和时钟一样是有刻度的&＃xff0c;每个刻度大小可以是100ms也可以是1ms&＃xff0c;如下图

上图的时间轮有6个刻度&＃xff0c;每个刻度大小是100ms&＃xff0c;也就是每过100ms会顺时针移动一个刻度&＃xff0c;走完一圈需要600ms&＃xff08;下面要介绍的HashedWheelTimer默认刻度数是512&＃xff0c;每个刻度大小默认是100ms&＃xff09;

工作原理&＃xff1a;每往时间轮提交一个延时任务&＃xff0c;会判断该任务的执行时间要放在哪个刻度上&＃xff0c;比如在时间轮启动后的第100ms&＃xff0c;提交了一个延时400ms执行的任务&＃xff0c;那么该任务应该放在刻度5上&＃xff0c;如果提交了一个延迟700ms执行的任务&＃xff0c;那么该任务会放在刻度2上&＃xff0c;并且会记录该任务还需要走一圈时间轮才能执行。时间轮每移动一个刻度&＃xff0c;就会执行当前刻度上的任务&＃xff0c;一个刻度上的任务可能会有多个。

因为HashedWheelTimer是基于时间轮的定时器&＃xff0c;所以接下来看一下HashedWheelTimer是如何实现的&＃xff1f;

三、HashedWheelTimer的相关组件

这里我们可以先看下HashedWheelTimer的UML图&＃xff0c;能够对相关组件先有个整体的认识&＃xff0c;如下

• Timer: 定时器接口&＃xff0c;提供提交延时任务newTimeout、停止定时器等方法

• HashedWheelTimer: 实现Timer接口&＃xff0c;内部包含工作线程Worker、时间轮wheel、延时任务队列timeouts、线程池taskExecutor等

• HashedWheelBucket&＃xff1a;上面的时间轮wheel是一个HashedWheelBucket数组&＃xff0c;每一个刻度对应一个HashedWheelBucket&＃xff0c;而每一个HashedWheelBucket内部是一个HashedWheelTimeout的双向链表&＃xff0c;如下图
  

• TimerTask: 延时任务接口&＃xff0c;内部只提供一个run方法用于执行

• Timeout: 对Timer、TimerTask的封装

• HashedWheelTimeout: 包含了任务的执行时间dealline、所需要的圈数remainingRounds、双向链表中上一个以及下一个HashedWheelTimeout、所在的HashedWheelBucket等

四、HashedWheelTimer的工作流程

大致工作流程如下图&＃xff1a;

从上图可以看到&＃xff0c;主要分为4步骤&＃xff0c;但是准确来说应该是有5步&＃xff1a;

1. 提交延时任务给HashedWheelTimer&＃xff0c;延时任务会先放到任务队列timeouts中
2. 工作线程Worker会从任务队列timeouts中获取任务
3. 将获取到的HashedWheelTimeout任务放到指定的HashedWheelBucket中
4. 取出当前刻度对应的HashedWheelBucket的所有HashedWheelTimeout来执行
5. 将刻度tick加1&＃xff0c;再回到第二步&＃xff0c;如此循环

五、源码解读

5.1 HahedWheelTimer的关键属性

关键属性如下&＃xff1a;

• Worker worker&＃xff1a;工作线程Worker
• int workerState&＃xff1a;工作线程状态
• long tickDuration&＃xff1a;刻度大小&＃xff0c;默认是100ms
• HashedWheelBucket[] wheel&＃xff1a;时间轮的每个刻度会对应一个HashedWheelBucket
• Queue timeouts&＃xff1a;任务队列
• Queue cancelledTimeouts&＃xff1a;已取消的任务队列
• AtomicLong pendingTimeouts&＃xff1a;正在处理的任务数
• Executor taskExecutor&＃xff1a;线程池&＃xff0c;用于执行任务
• long startTime&＃xff1a;定时器的启动时间

5.2 提交延时任务给HahedWheelTimer

通过newTimeout方法来提交延时任务&＃xff0c;newTimeout方法步骤如下&＃xff1a;

1. 启动工作线程Worker&＃xff0c;如果是首次启动&＃xff0c;设置启动时间startTime&＃xff0c;如果已启动&＃xff0c;则跳过
2. 计算延时任务的deadline&＃xff08;当前时间 &＃43; 延迟时间 - 启动时间startTime&＃xff09;&＃xff0c;用于判断后续放到时间轮的哪个HashedWheelBucket中
3. 将延时任务封装为HashedWheelTimeout&＃xff0c;并添加到任务队列timeouts中

结合源码来看&＃xff1a;

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {// 部分代码省略// 启动工作线程Workerstart();// 计算deadlinelong deadline &＃61; System.nanoTime() &＃43; unit.toNanos(delay) - startTime;if (delay > 0 && deadline < 0) {deadline &＃61; Long.MAX_VALUE;}// 封装为HashedWheelTimeout&＃xff0c;并添加到任务队列timeouts中HashedWheelTimeout timeout &＃61; new HashedWheelTimeout(this, task, deadline);timeouts.add(timeout);return timeout;
}


5.3 工作线程Worker运行的具体步骤

Worker类中有一个关键的属性tick&＃xff0c;代表相对于定时器的启动时间startTime&＃xff0c;当前已经走到了哪个刻度&＃xff0c;tick只会一直往上递增&＃xff0c;初始值为0

具体步骤如下&＃xff1a;

1. 等到下一个刻度来临&＃xff0c;即当前时间 > 当前刻度tick的结束时间
   a) 计算当前刻度tick的结束时间&＃xff0c;比如Worker刚启动&＃xff0c;当前刻度tick为0&＃xff0c;那么刻度tick的结束时间 &＃61; tickDuration * (tick &＃43; 1)&＃xff0c;即100ms
   b) 判断当前时间&＃xff08;相对于启动时间startTime&＃xff09;是否大于当前刻度的结束时间&＃xff0c;如果大于&＃xff0c;说明当前时间已经过了当前刻度的结束时间&＃xff0c;开始准备处理当前刻度的所有任务。如果小于&＃xff0c;说明当前时间还没到当前刻度的结束时间&＃xff0c;主动sleep一段时间后继续判断&＃xff0c;直到当前时间大于当前刻度的结束时间。
2. 从任务队列timeouts中获取任务&＃xff0c;将延时任务的deadline除以tickDuration&＃xff0c;计算出该任务的总刻度数以及还需要的圈数&＃xff0c;通过 **总刻度数 & &＃xff08;wheel.length -1 &＃xff09;**来算出放在哪个HashedWheelBucket中&＃xff08;比如算出A任务的总刻度数 &＃61; 1026&＃xff0c;当前刻度 &＃61; 25&＃xff0c;时间轮的刻度有512个&＃xff0c;那么算出还需要的圈数是1【如果当前刻度 &＃61; 1&＃xff0c;那么还需要的圈数会是2】&＃xff0c;放在下标为2的HashedWheelBucket中&＃xff09;
3. 获取当前刻度对应的HashedWheelBucket&＃xff0c;从head开始逐一遍历任务链表&＃xff0c;如果延时任务的所需圈数为0&＃xff0c;开始执行任务&＃xff0c;否则所需圈数减1。
4. 刻度tick加1&＃xff0c;回到第一步&＃xff0c;如此循环

结合源码来看

public void run() {// 初始化定时器的启动时间startTimestartTime &＃61; System.nanoTime();startTimeInitialized.countDown();do {// 1、等到下一个刻度来临&＃xff0c;即当前时间 > 当前刻度tick的结束时间final long deadline &＃61; waitForNextTick();if (deadline > 0) {// 获取当前刻度tick对应的HashedWheelBucketint idx &＃61; (int) (tick & mask);processCancelledTasks();HashedWheelBucket bucket &＃61; wheel[idx];// 2、从任务队列timeouts中获取任务&＃xff0c;并将任务放入到对应的HashedWheelBucket中 transferTimeoutsToBuckets();// 3、执行当前刻度tick对应的HashedWheelBucket中的所有任务bucket.expireTimeouts(deadline);// 4、将当前刻度tick加1tick&＃43;&＃43;;}} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) &＃61;&＃61; WORKER_STATE_STARTED);// 部分代码省略
}


这里我们先来看下第一步waitForNextTick方法的具体实现

private long waitForNextTick() {long deadline &＃61; tickDuration * (tick &＃43; 1);for (;;) {// 相对于startTime的当前时间final long currentTime &＃61; System.nanoTime() - startTime;long sleepTimeMs &＃61; (deadline - currentTime &＃43; 999999) / 1000000;// 如果当前刻度tick的结束时间 <当前时间&＃xff0c;说明当前时间已经过了当前刻度的结束时间&＃xff0c;直接返回当前时间if (sleepTimeMs <&＃61; 0) {if (currentTime &＃61;&＃61; Long.MIN_VALUE) {return -Long.MAX_VALUE;} else {return currentTime;}}// 否则主动sleep一段时间&＃xff0c;上面的条件成立try {Thread.sleep(sleepTimeMs);} catch (InterruptedException ignored) {if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) &＃61;&＃61; WORKER_STATE_SHUTDOWN) {return Long.MIN_VALUE;}}}
}


接着看下第二步的transferTimeoutsToBuckets方法&＃xff0c;如下

private void transferTimeoutsToBuckets() {// 这里一次最多从队列里获取100000个任务for (int i &＃61; 0; i < 100000; i&＃43;&＃43;) {HashedWheelTimeout timeout &＃61; timeouts.poll();if (timeout &＃61;&＃61; null) {// 代表队列里已经没有任务&＃xff0c;直接返回break;}if (timeout.state() &＃61;&＃61; HashedWheelTimeout.ST_CANCELLED) {continue;}// 计算总刻度数 &＃61; 延时任务的deadline / 刻度大小long calculated &＃61; timeout.deadline / tickDuration;// 计算还需要的圈数 &＃61; 总刻度数 - 当前刻度 / 时间轮的刻度数timeout.remainingRounds &＃61; (calculated - tick) / wheel.length;final long ticks &＃61; Math.max(calculated, tick);// 计算放在哪个下标int stopIndex &＃61; (int) (ticks & mask);HashedWheelBucket bucket &＃61; wheel[stopIndex];// 将该任务放入到对应的HashedWheelBucket中bucket.addTimeout(timeout);}
}


最后看下第三步bucket.expireTimeouts&＃xff0c;源码如下&＃xff1a;

public void expireTimeouts(long deadline) {HashedWheelTimeout timeout &＃61; head;// 处理该HashedWheelBucket的所有任务while (timeout !&＃61; null) {HashedWheelTimeout next &＃61; timeout.next;if (timeout.remainingRounds <&＃61; 0) {// 将任务从双向链表中移除next &＃61; remove(timeout);if (timeout.deadline <&＃61; deadline) {// 执行任务timeout.expire();} else {// The timeout was placed into a wrong slot. This should never happen.throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));}} else if (timeout.isCancelled()) {next &＃61; remove(timeout);} else {// 如果所需圈数 > 0&＃xff0c;则将其减1timeout.remainingRounds --;}timeout &＃61; next;}
}


至此&＃xff0c;工作线程Worker运行的具体步骤以及部分源码的解读就完成了

六、总结

HashedWheelTimer只是定时器的一种简单实现&＃xff0c;像java中常见的定时器还有Timer、ScheduledThreadPoolExecutor等&＃xff0c;从上面分析它的实现原理可知&＃xff0c;它无法应用于需要精确执行的场景&＃xff0c;但是在网络应用中&＃xff0c;IO任务的执行时间往往不需要精确&＃xff0c;所以它可以在任务较多、但任务不需要精确执行的场景下进行使用。

参考

【Netty】八、基于时间轮的定时器HashedWheelTimer