时刻轮（Time Wheel）是一种数据结构，用于完成依据时刻的事情调度体系。它被广泛使用于计算机网络、操作体系、分布式体系等领域中。

时刻轮概念

时刻轮的根本思想是将时刻划分为固定巨细的时刻段，并将这些时刻段组成一个环形结构。每个时刻段对应一个槽（Slot），槽中保存了在该时刻段内需要履行的事情列表。时刻轮按照固定的时刻距离（通常为1秒）逐个推动，当时刻轮的指针指向某个槽时，就履行该槽中一切事情。

长处

高效处理批量使命

时刻轮能够高效的使用线程资源来进行批量化调度，把大批量的调度使命全部都绑守时刻轮上，经过时刻轮进行一切使命的办理，触发以及运行。

降低时刻复杂度

时刻轮算法能够将插入和删去操作的时刻复杂度都降为O(1)，相较于JDK 供给的 java.util.Timer 和 DelayedQueue 等工具类，其底层完成使用的是堆这种数据结构，存取操作的复杂度都是 O(nlog(n))，无法支撑很多的守时使命。

缺陷

时刻精确度的问题

时刻轮调度器的时刻的精度或许不是很高，对于精度要求特别高的调度使命或许不太适合。因为时刻轮算法的精度取决于时刻段“指针”单元的最小粒度巨细，比如时刻轮的格子是一秒跳一次，那么调度精度小于一秒的使命就无法被时刻轮所调度。

能够选用多级时刻轮提高精度

宕机后无法恢复从头调度

时刻轮使命行列存储在内存中，没有做宕机备份，无法在宕机恢复后从头调度。

支撑宕机恢复需要做很多额外操作，能够参考携程QMQ完成

经典完成

DUBBO中时刻轮完成

Dubbo 的时刻轮完成坐落 dubbo-common 模块的 org.apache.dubbo.common.timer 包中

中心接口

首要有下面几个完成时刻轮

Timer

接口界说了守时器的根本行为

public interface Timer {
    //提交一个守时使命（TimerTask）并回来关联的 Timeout 目标
    Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
    Set<Timeout> stop();
    boolean isStop();
}

Timeout

界说了一个超时处理器的操作接口，由newTimeout办法回来

public interface Timeout {
    //界说了一个超时处理器的操作接口，由newTimeout办法回来
    Timer timer();
    //回来与此Timeout实例关联的TimerTask目标
    TimerTask task();
    boolean isExpired();
    boolean isCancelled();
    boolean cancel();
}

TimerTask

详细使命的基类,Timeout 目标与 TimerTask 目标一一对应，两者的联系类似于线程池回来的 Future 目标与提交到线程池中的使命目标之间的联系。经过 Timeout 目标，咱们不仅能够检查守时使命的状况，还能够操作守时使命（例如撤销关联的守时使命）

public interface TimerTask {
    void run(Timeout timeout) throws Exception;
}

HashedWheelTimeout

HashedWheelTimeout 是 Timeout 接口的仅有完成，是 HashedWheelTimer 的内部类。 HashedWheelTimeout 扮演了两个角色：

• 第一个，时刻轮中双向链表的节点，即守时使命 TimerTask 在 HashedWheelTimer 中的容器。
• 第二个，守时使命 TimerTask 提交到 HashedWheelTimer 之后回来的句柄（Handle），用于在时刻轮外部检查和控制守时使命。

关键字段和中心办法如下：

private static final class HashedWheelTimeout implements Timeout {
    //状况
    private static final int ST_INIT = 0;
    private static final int ST_CANCELLED = 1;
    private static final int ST_EXPIRED = 2;
    //状况更新
    private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
   @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization"})
    private volatile int state = ST_INIT;
    private final HashedWheelTimer timer;
    // 指实践被调度的使命
    private final TimerTask task;
    // 指守时使命履行的时刻。这个时刻是在创立 HashedWheelTimeout 时指定的，计算公式是：currentTime（创立 HashedWheelTimeout 的时刻） + delay（使命推迟时刻） - startTime（HashedWheelTimer 的发动时刻），时刻单位为纳秒。
    private final long deadline;
    //指当时使命剩下的时钟周期数。
    long remainingRounds;
    HashedWheelTimeout next;
    HashedWheelTimeout prev;
    HashedWheelBucket bucket;
    HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
        this.timer = timer;
        this.task = task;
        this.deadline = deadline;
    }
    @Override
    public Timer timer() {
        return timer;
    }
    @Override
    public TimerTask task() {
        return task;
    }
    @Override
    public boolean cancel() {
        // only update the state it will be removed from HashedWheelBucket on next tick.
        if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
            return false;
        }
        timer.cancelledTimeouts.add(this);
        return true;
    }
    void remove() {
        HashedWheelBucket bucket = this.bucket;
        if (bucket != null) {
            bucket.remove(this);
        } else {
            timer.pendingTimeouts.decrementAndGet();
        }
    }
    public boolean compareAndSetState(int expected, int state) {
        return STATE_UPDATER.compareAndSet(this, expected, state);
    }
    public int state() {
        return state;
    }
    @Override
    public boolean isCancelled() {
        return state() == ST_CANCELLED;
    }
    @Override
    public boolean isExpired() {
        return state() == ST_EXPIRED;
    }
    public void expire() {
        if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
            return;
        }
        try {
            task.run(this);
        } catch (Throwable t) {
            if (logger.isWarnEnabled()) {
                logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
            }
        }
    }
    @Override
    public String toString() {
        ...
    }
}

HashedWheelBucket

HashedWheelTimer的内部完成类，HashedWheelBucket 是时刻轮中的一个槽，时刻轮中的槽实践上就是一个用于缓存和办理双向链表的容器，双向链表中的每一个节点就是一个 HashedWheelTimeout 目标，也就关联了一个 TimerTask 守时使命。

HashedWheelBucket 持有双向链表的首尾两个节点，分别是 head 和 tail 两个字段，再加上每个 HashedWheelTimeout 节点均持有前驱和后继的引证，这样就能够正向或是逆向遍历整个双向链表了。

private static final class HashedWheelBucket {
    /**
     * Used for the linked-list datastructure
     */
    private HashedWheelTimeout head;
    private HashedWheelTimeout tail;
    /**
     * Add {@link HashedWheelTimeout} to this bucket.
     */
    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;
        }
    }
    /**
     * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
     */
    void expireTimeouts(long deadline) {
        HashedWheelTimeout timeout = head;
        // process all timeouts
        while (timeout != null) {
            HashedWheelTimeout next = timeout.next;
            if (timeout.remainingRounds <= 0) {
                next = remove(timeout);
                if (timeout.deadline <= 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 = remove(timeout);
            } else {
                timeout.remainingRounds--;
            }
            timeout = next;
        }
    }
    public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
        HashedWheelTimeout next = timeout.next;
        // remove timeout that was either processed or cancelled by updating the linked-list
        if (timeout.prev != null) {
            timeout.prev.next = next;
        }
        if (timeout.next != null) {
            timeout.next.prev = timeout.prev;
        }
        if (timeout == head) {
            // if timeout is also the tail we need to adjust the entry too
            if (timeout == tail) {
                tail = null;
                head = null;
            } else {
                head = next;
            }
        } else if (timeout == tail) {
            // if the timeout is the tail modify the tail to be the prev node.
            tail = timeout.prev;
        }
        // null out prev, next and bucket to allow for GC.
        timeout.prev = null;
        timeout.next = null;
        timeout.bucket = null;
        timeout.timer.pendingTimeouts.decrementAndGet();
        return next;
    }
    /**
     * Clear this bucket and return all not expired / cancelled {@link Timeout}s.
     */
    void clearTimeouts(Set<Timeout> set) {
        for (; ; ) {
            HashedWheelTimeout timeout = pollTimeout();
            if (timeout == null) {
                return;
            }
            if (timeout.isExpired() || timeout.isCancelled()) {
                continue;
            }
            set.add(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;
    }
}

HashedWheelTimer

时刻轮的仅有完成
HashedWheelTimer 会依据当时时刻轮指针选定对应的槽（HashedWheelBucket），从双向链表的头部开端迭代，对每个守时使命（HashedWheelTimeout）进行计算，归于当时时钟周期则取出运行，不归于则将其剩下的时钟周期数减一操作。
中心特点

//时刻轮的实例数
private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger();
private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean();
private static final int INSTANCE_COUNT_LIMIT = 64;
//状况更新原子修改
private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
        AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
//真实履行守时使命的逻辑封装这个 Runnable 目标中
private final Worker worker = new Worker();
//时刻轮内部真实履行守时使命的线程。
private final Thread workerThread;
private static final int WORKER_STATE_INIT = 0;
private static final int WORKER_STATE_STARTED = 1;
private static final int WORKER_STATE_SHUTDOWN = 2;
/**
 * 0 - init, 1 - started, 2 - shut down
 */
@SuppressWarnings({"unused", "FieldMayBeFinal"})
private volatile int workerState;
//时刻指针每次加 1 所代表的实践时刻，单位为纳秒
private final long tickDuration;
//该数组就是时刻轮的环形行列，每一个元素都是一个槽。当指守时刻轮槽数为 n 时，实践上会取大于且最靠近 n 的 2 的幂次方值。
private final HashedWheelBucket[] wheel;
//掩码， mask = wheel.length - 1，履行 ticks & mask 便能定位到对应的时钟槽
private final int mask;
private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
//timeouts 行列用于缓冲外部提交时刻轮中的守时使命
private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
//cancelledTimeouts 行列用于暂存撤销的守时使命 HashedWheelTimer 会在处理 HashedWheelBucket 的双向链表之前，先处理这两个行列中的数据。
private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();
//等候使命数
private final AtomicLong pendingTimeouts = new AtomicLong(0);
//最大使命数
private final long maxPendingTimeouts;
//开端时刻
private volatile long startTime;

结构办法

public HashedWheelTimer(
        ThreadFactory threadFactory,
        long tickDuration, TimeUnit unit, int ticksPerWheel,
        long maxPendingTimeouts) {
    if (threadFactory == null) {
        throw new NullPointerException("threadFactory");
    }
    if (unit == null) {
        throw new NullPointerException("unit");
    }
    if (tickDuration <= 0) {
        throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
    }
    if (ticksPerWheel <= 0) {
        throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
    }
    // Normalize ticksPerWheel to power of two and initialize the wheel.
    wheel = createWheel(ticksPerWheel);
    mask = wheel.length - 1;
    // Convert tickDuration to nanos.
    this.tickDuration = unit.toNanos(tickDuration);
    // Prevent overflow.
    if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
        throw new IllegalArgumentException(String.format(
                "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                tickDuration, Long.MAX_VALUE / wheel.length));
    }
    workerThread = threadFactory.newThread(worker);
    this.maxPendingTimeouts = maxPendingTimeouts;
    if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
            WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
        reportTooManyInstances();
    }
}
//创立槽
private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
    if (ticksPerWheel <= 0) {
        throw new IllegalArgumentException(
                "ticksPerWheel must be greater than 0: " + ticksPerWheel);
    }
    if (ticksPerWheel > 1073741824) {
        throw new IllegalArgumentException(
                "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
    }
    ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
    HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
    for (int i = 0; i < wheel.length; i++) {
        wheel[i] = new HashedWheelBucket();
    }
    return wheel;
}
//取ticksPerWheel往上最接近的2的幂次方
private static int normalizeTicksPerWheel(int ticksPerWheel) {
    int normalizedTicksPerWheel = ticksPerWheel - 1;
    normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 1;
    normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 2;
    normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 4;
    normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 8;
    normalizedTicksPerWheel |= normalizedTicksPerWheel >>> 16;
    return normalizedTicksPerWheel + 1;
}

发动时刻轮,结构函数中调用

public void start() {
    switch (WORKER_STATE_UPDATER.get(this)) {
        case WORKER_STATE_INIT:
            if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                workerThread.start();
            }
            break;
        case WORKER_STATE_STARTED:
            break;
        case WORKER_STATE_SHUTDOWN:
            throw new IllegalStateException("cannot be started once stopped");
        default:
            throw new Error("Invalid WorkerState");
    }
    // Wait until the startTime is initialized by the worker.
    while (startTime == 0) {
        try {
            startTimeInitialized.await();
        } catch (InterruptedException ignore) {
            // Ignore - it will be ready very soon.
        }
    }
}

中止时刻轮

public Set<Timeout> stop() {
    if (Thread.currentThread() == workerThread) {
        throw new IllegalStateException(
                HashedWheelTimer.class.getSimpleName() +
                        ".stop() cannot be called from " +
                        TimerTask.class.getSimpleName());
    }
    if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
        // workerState can be 0 or 2 at this moment - let it always be 2.
        if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
            INSTANCE_COUNTER.decrementAndGet();
        }
        return Collections.emptySet();
    }
    try {
        boolean interrupted = false;
        while (workerThread.isAlive()) {
            workerThread.interrupt();
            try {
                //等候100ms不一定履行完
                workerThread.join(100);
            } catch (InterruptedException ignored) {
                interrupted = true;
            }
        }
        if (interrupted) {
            //重设标记位
            Thread.currentThread().interrupt();
        }
    } finally {
        INSTANCE_COUNTER.decrementAndGet();
    }
    return worker.unprocessedTimeouts();
}

提交使命

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
    if (task == null) {
        throw new NullPointerException("task");
    }
    if (unit == null) {
        throw new NullPointerException("unit");
    }
    long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
    if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
        pendingTimeouts.decrementAndGet();
        throw new RejectedExecutionException("Number of pending timeouts ("
                + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
                + "timeouts (" + maxPendingTimeouts + ")");
    }
    start();
    // Add the timeout to the timeout queue which will be processed on the next tick.
    // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
    long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
    // Guard against overflow.
    if (delay > 0 && deadline < 0) {
        deadline = Long.MAX_VALUE;
    }
    HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
    timeouts.add(timeout);
    return timeout;
}

工作线程正常状况下是个死循环，共同履行

private final class Worker implements Runnable {
    private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
    private long tick;
    @Override
    public void run() {
        // Initialize the startTime.
        startTime = System.nanoTime();
        if (startTime == 0) {
            // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
            startTime = 1;
        }
        // Notify the other threads waiting for the initialization at start().
        startTimeInitialized.countDown();
        do {
            //等候下一个指针
            final long deadline = waitForNextTick();
            if (deadline > 0) {
                int idx = (int) (tick & mask);
                //处理撤销使命
                processCancelledTasks();
                HashedWheelBucket bucket =
                        wheel[idx];
                //把timeouts行列中的数据移到槽中，一次10w
                transferTimeoutsToBuckets();
                //履行槽中到期使命的run办法
                bucket.expireTimeouts(deadline);
                tick++;
            }
        } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
        // Fill the unprocessedTimeouts so we can return them from stop() method.
        for (HashedWheelBucket bucket : wheel) {
            bucket.clearTimeouts(unprocessedTimeouts);
        }
        for (; ; ) {
            HashedWheelTimeout timeout = timeouts.poll();
            if (timeout == null) {
                break;
            }
            if (!timeout.isCancelled()) {
                unprocessedTimeouts.add(timeout);
            }
        }
        processCancelledTasks();
    }
    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) {
                // all processed
                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);
        }
    }
    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);
                }
            }
        }
    }
    /**
     * calculate goal nanoTime from startTime and current tick number,
     * then wait until that goal has been reached.
     *
     * @return Long.MIN_VALUE if received a shutdown request,
     * current time otherwise (with Long.MIN_VALUE changed by +1)
     */
    private long waitForNextTick() {
        long deadline = tickDuration * (tick + 1);
        for (; ; ) {
            final long currentTime = System.nanoTime() - startTime;
            long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
            if (sleepTimeMs <= 0) {
                if (currentTime == Long.MIN_VALUE) {
                    return -Long.MAX_VALUE;
                } else {
                    return currentTime;
                }
            }
            if (isWindows()) {
                sleepTimeMs = sleepTimeMs / 10 * 10;
            }
            try {
                Thread.sleep(sleepTimeMs);
            } catch (InterruptedException ignored) {
                if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
                    return Long.MIN_VALUE;
                }
            }
        }
    }
    Set<Timeout> unprocessedTimeouts() {
        return Collections.unmodifiableSet(unprocessedTimeouts);
    }
}

使用场景

在 Dubbo 中，时刻轮并不直接用于周期性操作，而是只向时刻轮提交履行单次的守时使命，在上一次使命履行完成的时分，调用 newTimeout() 办法再次提交当时使命，这样就会在下个周期履行该使命。即便在使命履行过程中呈现了 GC、I/O 阻塞等状况，导致使命推迟或卡住，也不会有相同的使命源源不断地提交进来，导致使命堆积。

Dubbo 中对时刻轮的使用首要体现在如下两个方面：

• 失利重试， 例如，Provider 向注册中心进行注册失利时的重试操作，或是 Consumer 向注册中心订阅时的失利重试等。

//FailbackRegistry
private void addFailedRegistered(URL url) {
 FailedRegisteredTask oldOne = failedRegistered.get(url);
 if (oldOne != null) {
     return;
 }
 FailedRegisteredTask newTask = new FailedRegisteredTask(url, this);
 oldOne = failedRegistered.putIfAbsent(url, newTask);
 if (oldOne == null) {
     // never has a retry task. then start a new task for retry.
     retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
 }
}

• 周期性守时使命， 例如，守时发送心跳恳求，恳求超时的处理，或是网络连接断开后的重连机制。

总结

• DUBBO中时刻轮使用的较为简略，未考虑的大跨度时刻问题和重启恢复，应该首要是其使用场景用不到这些
• 不过其对时刻轮中时刻加了剩下履行轮数优化remainingRounds，每转一圈会减一
• 详细的使命都是在一个work线程中履行，大批量使命履行时或许会导致履行时刻精度不准

剩下部分请见算法技巧-时刻轮（二）

参考
1.05 海量守时使命，一个时刻轮搞定
2.延时使命一锅端