首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >面试Java后端却问我时间轮算法,面试官没想到我看过Dubbo源码!

面试Java后端却问我时间轮算法,面试官没想到我看过Dubbo源码!

作者头像
JavaEdge
修改2026-06-22 14:24:20
修改2026-06-22 14:24:20
7560
举报
文章被收录于专栏:JavaEdgeJavaEdge

1 定时任务的解决方案

Netty、Quartz、Kafka 以及 Linux 都有定时任务功能。

  • 常规的JDK 的 java.util.Timer 和 DelayedQueue 等工具类,可实现简单的定时任务,底层用的是数据结构,存取复杂度$O(nlog(n))$,无法支撑海量定时任务
  • 但在定时任务量大、性能要求高的场景,为将任务存取及取消操作时间复杂度降为$O(1)$,会使用时间轮

2 时间轮模型

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

Hashed Timing Wheel 结构示意图

3 适用场景

  • 故障恢复
  • 流量控制
  • 调度算法
  • 控制网络中的数据包生命周期

计时器维护代价高,若:

  • 处理器在每个时钟滴答声中都会中断
  • 使用精细粒度计时器
  • 未完成的计时器很多

需要高效的定时器算法,以减少总体中断的开销。

单层时间轮的容量和精度都有限,对精度要求特别高、时间跨度特别大或是海量定时任务需要调度的场景,通常使用多级时间轮以及持久化存储与时间轮结合的方案。

4 模型和性能指标

4.1 模型中的规则

  • 客户端调用: - START_TIMER(时间间隔,Request_ID,Expiry_Action) - STOP_TIMER(Request_ID)
  • 计时器tick调用: - PER_TICK_BOOKKEEPING - EXPIRY_PROCESSING

4.2 性能指标

  • 空间 数据结构使用的内存
  • 延迟 开始和结束上述任何例程所需的时间5 Dubbo的时间轮实现

Dubbo时间轮实现位于 dubbo-common 模块的 org.apache.dubbo.common.timer 包中,分析时间轮涉及的核心接口和实现。

5.1 核心接口

代码语言:mermaid
复制
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()
TimerTask

Dubbo的所有定时任务都要实现 TimerTask 接口。只定义 run() 方法,入参是 Timeout 接口对象。

代码语言:java
复制
/**
 * 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

Timeout 对象与 TimerTask 对象一一对应,类似线程池返回的 Future 对象与提交到线程池中的任务对象之间的关系。

通过 Timeout 对象,不仅能查看定时任务的状态,还能操作定时任务(例如取消关联的定时任务)。

代码语言:java
复制
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();
}
Timer 接口

定义了定时器的基本行为,核心是 newTimeout() :提交一个定时任务(TimerTask)并返回关联的 Timeout 对象,类似于向线程池提交任务。

代码语言:java
复制
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();
}

6 HashedWheelTimeout

Timeout 接口的唯一实现,HashedWheelTimer 的内部类,扮演如下角色:

  • 时间轮中双向链表的节点,即定时任务 TimerTask 在 HashedWheelTimer 中的容器
  • 定时任务 TimerTask 提交到 HashedWheelTimer 后返回的句柄(Handle),用于在时间轮外部查看和控制定时任务

6.1 核心字段

通过双向链表被用来在HashedWheelTimerBucket链timeouts(定时任务),由于只在WorkerThread上行动,没有必要进行同步/volatile。

代码语言:java
复制
HashedWheelTimeout next;
HashedWheelTimeout prev;

task:

代码语言:java
复制
 // 实际被调度的任务
private final TimerTask task;

deadline,定时任务执行的时间。在创建 HashedWheelTimeout 时指定。计算公式:

currentTime(创建 HashedWheelTimeout 的时间) + delay(任务延迟时间) - startTime(HashedWheelTimer 的启动时间),ns

代码语言:java
复制
private final long deadline;

state,定时任务当前所处状态

代码语言:java
复制
// 抑制以下警告:
//   "unused"            → 未使用
//   "FieldMayBeFinal"   → 字段可以声明为 final
//   "RedundantFieldInitialization" → 字段初始化是冗余的(因为 int 默认值就是 0)
@SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
private volatile int state = ST_INIT;
// volatile:保证多线程可见性
// state:状态字段,初始值为 ST_INIT(初始状态)

可选状态:

代码语言:java
复制
private static final int ST_INIT = 0;
private static final int ST_CANCELLED = 1;
private static final int ST_EXPIRED = 2;

STATE_UPDATER 字段,用于实现 state 状态变更的原子性:

代码语言:java
复制
private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
                AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");

remainingRounds,当前任务剩余的时钟周期数。时间轮所能表示的时间长度有限,在任务到期时间与当前时刻的时间差,超过时间轮单圈能表示时长,就出现套圈,需要该字段值表示剩余的时钟周期。

代码语言:java
复制
long remainingRounds;

核心API

代码语言:java
复制
// 当且仅当与此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();
    }
}

HashedWheelBucket

时间轮中的一个槽,实际上就是一个用于缓存和管理双向链表的容器,双向链表中的每一个节点就是一个 HashedWheelTimeout 对象,也就关联了一个 TimerTask 定时任务。

HashedWheelBucket 持有双向链表的首尾两个节点 - head 和 tail,再加上每个 HashedWheelTimeout 节点均持有前驱和后继引用,即可正、逆向遍历整个链表。

核心API

代码语言:java
复制
/**
 * 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;
    }
}
代码语言:java
复制
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 字段(剩余时钟周期数)减一。

HashedWheelTimer

Timer 接口的实现,通过时间轮算法实现了一个定时器。

职能

根据当前时间轮指针选定对应 HashedWheelBucket 槽,从链表头部开始迭代,计算每个 HashedWheelTimeout 定时任务:

  • 属于当前时钟周期则取出运行
  • 不属于则将其剩余的时钟周期数减一

核心域

代码语言:java
复制
// 时间轮当前所处状态,三个可选值,由 `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 的双向链表前,先处理这俩队列的数据:

代码语言:java
复制
// 缓冲外部提交时间轮中的定时任务
private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();

// 暂存取消的定时任务
private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();
代码语言:java
复制
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();

newTimeout()

提交定时任务,在定时任务进入到 timeouts 队列之前会先调用 start() 方法启动时间轮,其中会完成下面两个关键步骤:

  1. 确定时间轮的 startTime 字段
  2. 启动 workerThread 线程,开始执行 worker 任务。

之后根据 startTime 计算该定时任务的 deadline,最后才能将定时任务封装成 HashedWheelTimeout 并添加到 timeouts 队列。

4 时间轮指针一次转动的执行流程

HashedWheelTimer$Worker.run()

1、时间轮指针转动,时间轮周期开始

2、清理用户主动取消的定时任务,这些定时任务在用户取消时,记录到 cancelledTimeouts 队列中。在每次指针转动的时候,时间轮都会清理该队列

代码语言:java
复制
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 队列中的定时任务转移到时间轮中对应的槽

代码语言:java
复制
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、检测时间轮的状态。如果时间轮处于运行状态,则循环执行上述步骤,不断执行定时任务。

代码语言:java
复制
while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED)

若时间轮处于停止状态,则执行下面的步骤,获取到未被执行的定时任务并加入 unprocessedTimeouts 队列

代码语言:java
复制
// 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 队列中用户主动取消的定时任务。

代码语言:java
复制
processCancelledTasks();

定时任务应用

并不直接用于周期性操作,而是只向时间轮提交执行单次的定时任务,在上一次任务执行完成的时候,调用 newTimeout() 方法再次提交当前任务,这样就会在下个周期执行该任务。即使在任务执行过程中出现了 GC、I/O 阻塞等情况,导致任务延迟或卡住,也不会有同样的任务源源不断地提交进来,导致任务堆积。

Dubbo 时间轮应用主要在:

  1. 失败重试, 如Provider 向注册中心进行注册失败时的重试或Consumer 向注册中心订阅时的失败重试
  2. 周期性定时任务, 如定期发送心跳请求,请求超时的处理,或是网络连接断开后的重连机制
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2020/10/14 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 1 定时任务的解决方案
  • 2 时间轮模型
    • Hashed Timing Wheel 结构示意图
  • 3 适用场景
  • 4 模型和性能指标
    • 4.1 模型中的规则
    • 4.2 性能指标
    • 5.1 核心接口
    • Timeout
  • 6 HashedWheelTimeout
    • 6.1 核心字段
    • 核心API
  • HashedWheelBucket
    • 核心API
  • HashedWheelTimer
    • 职能
    • 核心域
    • newTimeout()
  • 4 时间轮指针一次转动的执行流程
    • 定时任务应用
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档