
Netty、Quartz、Kafka 以及 Linux 都有定时任务功能。
一种高效批量管理定时任务的调度模型。时间轮一般实现成一个环形结构,似时钟,分为很多槽,一个槽代表一个时间间隔,每个槽内部使用双向链表存储定时任务。指针周期性跳动,跳动到一个槽位,就执行该槽位里的定时任务。

计时器维护代价高,若:
需要高效的定时器算法,以减少总体中断的开销。
单层时间轮的容量和精度都有限,对精度要求特别高、时间跨度特别大或是海量定时任务需要调度的场景,通常使用多级时间轮以及持久化存储与时间轮结合的方案。
Dubbo时间轮实现位于 dubbo-common 模块的 org.apache.dubbo.common.timer 包中,分析时间轮涉及的核心接口和实现。
classDiagram
class Timer {
+newTimeout(TimerTask, long, TimeUnit) Timeout
+stop() Set~Timeout~
+isStop() boolean
}
class Timeout {
+timer() Timer
+task() TimerTask
+isExpired() boolean
+isCancelled() boolean
+cancel() boolean
}
class TimerTask {
+run(Timeout) void
}
Timer ..> Timeout : creates / returns
Timeout ..> Timer : references
Timeout ..> TimerTask : references
TimerTask ..> Timeout : uses in run()Dubbo的所有定时任务都要实现 TimerTask 接口。只定义 run() 方法,入参是 Timeout 接口对象。
/**
* A task which is executed after the delay specified with
* {@link Timer#newTimeout(TimerTask, long, TimeUnit)} (TimerTask, long, TimeUnit)}.
*/
public interface TimerTask {
/**
* Executed after the delay specified with
* {@link Timer#newTimeout(TimerTask, long, TimeUnit)}.
*
* @param timeout a handle which is associated with this task
*/
void run(Timeout timeout) throws Exception;
}Timeout 对象与 TimerTask 对象一一对应,类似线程池返回的 Future 对象与提交到线程池中的任务对象之间的关系。
通过 Timeout 对象,不仅能查看定时任务的状态,还能操作定时任务(例如取消关联的定时任务)。
package org.apache.dubbo.common.timer;
/**
* A handle associated with a {@link TimerTask} that is returned by a
* {@link Timer}.
*/
public interface Timeout {
/**
* Returns the {@link Timer} that created this handle.
*/
Timer timer();
/**
* Returns the {@link TimerTask} which is associated with this handle.
*/
TimerTask task();
/**
* Returns {@code true} if and only if the {@link TimerTask} associated
* with this handle has been expired.
*/
boolean isExpired();
/**
* Returns {@code true} if and only if the {@link TimerTask} associated
* with this handle has been cancelled.
*/
boolean isCancelled();
/**
* Attempts to cancel the {@link TimerTask} associated with this handle.
* If the task has been executed or cancelled already, it will return with
* no side effect.
*
* @return True if the cancellation completed successfully, otherwise false
*/
boolean cancel();
}定义了定时器的基本行为,核心是 newTimeout() :提交一个定时任务(TimerTask)并返回关联的 Timeout 对象,类似于向线程池提交任务。
package org.apache.dubbo.common.timer;
/**
* Schedules {@link TimerTask}s for one-time future execution in a background
* thread.
*/
public interface Timer {
/**
* Schedules the specified {@link TimerTask} for one-time execution after
* the specified delay.
*
* @return a handle which is associated with the specified task
* @throws IllegalStateException if this timer has been {@linkplain #stop() stopped} already
* @throws RejectedExecutionException if the pending timeouts are too many and creating new timeout
* can cause instability in the system.
*/
Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
/**
* Releases all resources acquired by this {@link Timer} and cancels all
* tasks which were scheduled but not executed yet.
*
* @return the handles associated with the tasks which were canceled by
* this method
*/
Set<Timeout> stop();
/**
* the timer is stop
*
* @return true for stop
*/
boolean isStop();
}Timeout 接口的唯一实现,HashedWheelTimer 的内部类,扮演如下角色:
通过双向链表被用来在HashedWheelTimerBucket链timeouts(定时任务),由于只在WorkerThread上行动,没有必要进行同步/volatile。
HashedWheelTimeout next;
HashedWheelTimeout prev;task:
// 实际被调度的任务
private final TimerTask task;deadline,定时任务执行的时间。在创建 HashedWheelTimeout 时指定。计算公式:
currentTime(创建 HashedWheelTimeout 的时间) + delay(任务延迟时间) - startTime(HashedWheelTimer 的启动时间),ns
private final long deadline;state,定时任务当前所处状态
// 抑制以下警告:
// "unused" → 未使用
// "FieldMayBeFinal" → 字段可以声明为 final
// "RedundantFieldInitialization" → 字段初始化是冗余的(因为 int 默认值就是 0)
@SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
private volatile int state = ST_INIT;
// volatile:保证多线程可见性
// state:状态字段,初始值为 ST_INIT(初始状态)可选状态:
private static final int ST_INIT = 0;
private static final int ST_CANCELLED = 1;
private static final int ST_EXPIRED = 2;STATE_UPDATER 字段,用于实现 state 状态变更的原子性:
private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");remainingRounds,当前任务剩余的时钟周期数。时间轮所能表示的时间长度有限,在任务到期时间与当前时刻的时间差,超过时间轮单圈能表示时长,就出现套圈,需要该字段值表示剩余的时钟周期。
long remainingRounds;// 当且仅当与此handle关联的TimerTask已被取消时,返回true
// 用于判断与该超时句柄关联的定时任务是否已被取消
@Override
public boolean isCancelled() {
return state() == ST_CANCELLED;
}
/**
* 当且仅当与此句柄关联的TimerTask已过期(超时)时,返回true
*/
@Override
public boolean isExpired() {
return state() == ST_EXPIRED;
}
// 检查当前 HashedWheelTimeout 状态
public int state() {
return state;
}
@Override
public boolean cancel() {
// only update the state it will be removed from HashedWheelBucket on next tick.
// 仅更新状态,下次将其从HashedWheelBucket中删除。
if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
return false;
}
// 如果应该取消任务,我们将其放入另一个队列,该队列将在每个刻度上处理
// 因此,这意味着我们的GC延迟为max。 1个刻度持续时间就足够了
// 这样,我们就可以再次使用我们的 MpscLinkedQueue,从而尽可能减少锁/开销。
// If a task should be canceled we put this to another queue which will be processed on each tick.
// So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
// we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
timer.cancelledTimeouts.add(this);
return true;
}
public void expire() {
// 当任务到期时,将当前 HashedWheelTimeout 设为 EXPIRED 态
if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
return;
}
try {
// 然后调用其 run() 方法执行定时任务。
task.run(this);
}
...
}
/** 将当前 HashedWheelTimeout 从时间轮中删除。 */
void remove() {
HashedWheelBucket bucket = this.bucket;
if (bucket != null) {
bucket.remove(timeout: this);
} else {
timer.pendingTimeouts.decrementAndGet();
}
}时间轮中的一个槽,实际上就是一个用于缓存和管理双向链表的容器,双向链表中的每一个节点就是一个 HashedWheelTimeout 对象,也就关联了一个 TimerTask 定时任务。
HashedWheelBucket 持有双向链表的首尾两个节点 - head 和 tail,再加上每个 HashedWheelTimeout 节点均持有前驱和后继引用,即可正、逆向遍历整个链表。
/**
* Add {@link HashedWheelTimeout} to this bucket.
*/
public void addTimeout(HashedWheelTimeout timeout) {
assert timeout.bucket == null;
timeout.bucket = this;
if (head == null) {
head = tail = timeout;
} else {
tail.next = timeout;
timeout.prev = tail;
tail = timeout;
}
}private HashedWheelTimeout pollTimeout() {
HashedWheelTimeout head = this.head;
if (head == null) {
return null;
}
HashedWheelTimeout next = head.next;
if (next == null) {
tail = this.head = null;
} else {
this.head = next;
next.prev = null;
}
// null out prev and next to allow for GC.
head.next = null;
head.prev = null;
head.bucket = null;
return head;
}remove():从双向链表中移除指定的 HashedWheelTimeout 节点。
clearTimeouts():循环调用 pollTimeout() 方法处理整个双向链表,并返回所有未超时或者未被取消的任务。
expireTimeouts():遍历双向链表中的全部 HashedWheelTimeout 节点。 在处理到期的定时任务时,会通过 remove() 方法取出,并调用其 expire() 方法执行;对于已取消的任务,通过 remove() 方法取出后直接丢弃;对于未到期的任务,会将 remainingRounds 字段(剩余时钟周期数)减一。
Timer 接口的实现,通过时间轮算法实现了一个定时器。
根据当前时间轮指针选定对应 HashedWheelBucket 槽,从链表头部开始迭代,计算每个 HashedWheelTimeout 定时任务:
// 时间轮当前所处状态,三个可选值,由 `AtomicIntegerFieldUpdater` 实现其原子地修改
@SuppressWarnings({ "unused", "FieldMayBeFinal" })
private volatile int workerState; // 0 - init, 1 - started, 2 - shut down
// 当前时间轮的启动时间,提交到该时间轮的定时任务的 deadline 字段值均以该时间戳为起点进行计算。
private volatile long startTime;
// 时间轮环形队列,每个元素都是一个槽。当指定时间轮槽数为 n 时,会向上取最靠近 n 的 2 次幂值
private final HashedWheelBucket[] wheel;HashedWheelTimer 会在处理 HashedWheelBucket 的双向链表前,先处理这俩队列的数据:
// 缓冲外部提交时间轮中的定时任务
private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
// 暂存取消的定时任务
private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();private final class Worker implements Runnable {
private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
// 位于 `HashedWheelTimer$Worker` ,时间轮的指针,步长为 1 的单调递增计数器
private long tick;
// 掩码, `mask = wheel.length - 1`,执行 `ticks & mask` 便能定位到对应的时钟槽
private final int mask;
// 时间指针每次加 1 所代表的实际时间,单位为纳秒。
private final long tickDuration;
// 当前时间轮剩余的定时任务总数
private final AtomicLong pendingTimeouts = new AtomicLong(0);
// 时间轮内部真正执行定时任务的线程
private final Thread workerThread;
// 真正执行定时任务的逻辑封装这个 Runnable 对象
private final Worker worker = new Worker();提交定时任务,在定时任务进入到 timeouts 队列之前会先调用 start() 方法启动时间轮,其中会完成下面两个关键步骤:
之后根据 startTime 计算该定时任务的 deadline,最后才能将定时任务封装成 HashedWheelTimeout 并添加到 timeouts 队列。
HashedWheelTimer$Worker.run():
1、时间轮指针转动,时间轮周期开始
2、清理用户主动取消的定时任务,这些定时任务在用户取消时,记录到 cancelledTimeouts 队列中。在每次指针转动的时候,时间轮都会清理该队列
private void processCancelledTasks() {
for (; ; ) {
HashedWheelTimeout timeout = cancelledTimeouts.poll();
if (timeout == null) {
// all processed
break;
}
try {
timeout.remove();
} catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("An exception was thrown while process a cancellation task", t);
}
}
}
} 3、将缓存在 timeouts 队列中的定时任务转移到时间轮中对应的槽
private void transferTimeoutsToBuckets() {
// transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
// adds new timeouts in a loop.
for (int i = 0; i < 100000; i++) {
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
// 都处理完了
break;
}
if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
// Was cancelled in the meantime.
continue;
}
long calculated = timeout.deadline / tickDuration;
timeout.remainingRounds = (calculated - tick) / wheel.length;
// Ensure we don't schedule for past.
final long ticks = Math.max(calculated, tick);
int stopIndex = (int) (ticks & mask);
HashedWheelBucket bucket = wheel[stopIndex];
bucket.addTimeout(timeout);
}
}4、根据当前指针定位对应槽,处理该槽位的双向链表中的定时任务
5、检测时间轮的状态。如果时间轮处于运行状态,则循环执行上述步骤,不断执行定时任务。
while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED)若时间轮处于停止状态,则执行下面的步骤,获取到未被执行的定时任务并加入 unprocessedTimeouts 队列
// Fill the unprocessedTimeouts so we can return them from stop() method.
// 遍历时间轮中每个槽位,并调用clearTimeouts()
for (HashedWheelBucket bucket : wheel) {
bucket.clearTimeouts(unprocessedTimeouts);
}
// 对 timeouts 队列中未被加入槽中循环调用poll()
for (; ; ) {
HashedWheelTimeout timeout = timeouts.poll();
if (timeout == null) {
break;
}
if (!timeout.isCancelled()) {
unprocessedTimeouts.add(timeout);
}
}6、最后再次清理 cancelledTimeouts 队列中用户主动取消的定时任务。
processCancelledTasks();并不直接用于周期性操作,而是只向时间轮提交执行单次的定时任务,在上一次任务执行完成的时候,调用 newTimeout() 方法再次提交当前任务,这样就会在下个周期执行该任务。即使在任务执行过程中出现了 GC、I/O 阻塞等情况,导致任务延迟或卡住,也不会有同样的任务源源不断地提交进来,导致任务堆积。
Dubbo 时间轮应用主要在: