时间轮 + Dubbo + HashedWheelTimer

梦里梦外; 2023-10-01 05:20 15阅读 0赞

#### Dubbo 时间轮算法简析 HashedWheelTimer源码及应用\* ####

*  一、Dubbo时间轮简要结构图
 *   *  1.时间轮结构
     *  2.项目结构（Dubbo源码取了3.0.2.1版本）
 *  二、源码分析（针对主要方法加了个人理解和翻译）
 *   *  1.定时器实现类：HashedWheelTimer
     *  2.任务处理接口类：Timeout
     *  3.定时器接口类：Timer
     *  4.任务接口类：TimerTask
 *  三、调用：
 *   *  1.测试接口：
     *  2.调试效果：

## 一、Dubbo时间轮简要结构图 ##

### 1.时间轮结构 ###

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16]

### 2.项目结构（Dubbo源码取了3.0.2.1版本） ###

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_18_color_FFFFFF_t_70_g_se_x_16]

## 二、源码分析（针对主要方法加了个人理解和翻译） ##

### 1.定时器实现类：HashedWheelTimer ###

/*
     * Copyright 2012 The Netty Project
     *
     * The Netty Project licenses this file to you under the Apache License,
     * version 2.0 (the "License"); you may not use this file except in compliance
     * with the License. You may obtain a copy of the License at:
     *
     *   http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
     * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
     * License for the specific language governing permissions and limitations
     * under the License.
     */
    
    package com.demo.config.wheel;
    
    import com.mchange.v1.lang.ClassUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.*;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicBoolean;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
    import java.util.concurrent.atomic.AtomicLong;
    
    /**
     * A {@link Timer} optimized for approximated I/O timeout scheduling.
     *
     * <h3>Tick Duration</h3>
     * <p>
     * As described with 'approximated', this timer does not execute the scheduled
     * {@link TimerTask} on time.  {@link HashedWheelTimer}, on every tick, will
     * check if there are any {@link TimerTask}s behind the schedule and execute
     * them.
     * <p>
     * You can increase or decrease the accuracy of the execution timing by
     * specifying smaller or larger tick duration in the constructor.  In most
     * network applications, I/O timeout does not need to be accurate.  Therefore,
     * the default tick duration is 100 milliseconds and you will not need to try
     * different configurations in most cases.
     *
     * <h3>Ticks per Wheel (Wheel Size)</h3>
     * <p>
     * {@link HashedWheelTimer} maintains a data structure called 'wheel'.
     * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash
     * function is 'dead line of the task'.  The default number of ticks per wheel
     * (i.e. the size of the wheel) is 512.  You could specify a larger value
     * if you are going to schedule a lot of timeouts.
     *
     * <h3>Do not create many instances.</h3>
     * <p>
     * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and
     * started.  Therefore, you should make sure to create only one instance and
     * share it across your application.  One of the common mistakes, that makes
     * your application unresponsive, is to create a new instance for every connection.
     *
     * <h3>Implementation Details</h3>
     * <p>
     * {@link HashedWheelTimer} is based on
     * <a href="http://cseweb.ucsd.edu/users/varghese/">George Varghese</a> and
     * Tony Lauck's paper,
     * <a href="http://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed
     * and Hierarchical Timing Wheels: data structures to efficiently implement a
     * timer facility'</a>.  More comprehensive slides are located
     * <a href="http://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here</a>.
     * @author
     */
    public class HashedWheelTimer implements Timer {
        
    
        /**
         * may be in spi?
         */
        public static final String NAME = "hased";
    
        private static final Logger logger = LoggerFactory.getLogger(HashedWheelTimer.class);
    
        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");
    
        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;
    
        /**
         * tickDuration 步长
         * wheel 轮子
         * mask 轮子格数-1
         * startTimeInitialized 线程计数器，每当一个线程执行完毕计数器值-1，当值为0时，表示所有线程都执行完毕，然后在闭锁上等待的线程就可以恢复工作了
         * timeouts 任务队列
         * cancelledTimeouts 已取消的任务队列
         * pendingTimeouts 剩余任务总数
         */
        private final long tickDuration;
        private final HashedWheelBucket[] wheel;
        private final int mask;
        private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
        private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
        private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();
        private final AtomicLong pendingTimeouts = new AtomicLong(0);
        private final long maxPendingTimeouts;
    
        private volatile long startTime;
    
        /**
         * Creates a new timer with the default thread factory
         * ({@link Executors#defaultThreadFactory()}), default tick duration, and
         * default number of ticks per wheel.
         */
        public HashedWheelTimer() {
        
            this(Executors.defaultThreadFactory());
        }
    
        /**
         * Creates a new timer with the default thread factory
         * ({@link Executors#defaultThreadFactory()}) and default number of ticks
         * per wheel.
         *
         * @param tickDuration the duration between tick
         * @param unit         the time unit of the {@code tickDuration}
         * @throws NullPointerException     if {@code unit} is {@code null}
         * @throws IllegalArgumentException if {@code tickDuration} is <= 0
         */
        public HashedWheelTimer(long tickDuration, TimeUnit unit) {
        
            this(Executors.defaultThreadFactory(), tickDuration, unit);
        }
    
        /**
         * Creates a new timer with the default thread factory
         * ({@link Executors#defaultThreadFactory()}).
         *
         * @param tickDuration  the duration between tick
         * @param unit          the time unit of the {@code tickDuration}
         * @param ticksPerWheel the size of the wheel
         * @throws NullPointerException     if {@code unit} is {@code null}
         * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
         */
        public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
        
            this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
        }
    
        /**
         * Creates a new timer with the default tick duration and default number of
         * ticks per wheel.
         *
         * @param threadFactory a {@link ThreadFactory} that creates a
         *                      background {@link Thread} which is dedicated to
         *                      {@link TimerTask} execution.
         * @throws NullPointerException if {@code threadFactory} is {@code null}
         */
        public HashedWheelTimer(ThreadFactory threadFactory) {
        
            this(threadFactory, 100, TimeUnit.MILLISECONDS);
        }
    
        /**
         * Creates a new timer with the default number of ticks per wheel.
         *
         * @param threadFactory a {@link ThreadFactory} that creates a
         *                      background {@link Thread} which is dedicated to
         *                      {@link TimerTask} execution.
         * @param tickDuration  the duration between tick
         * @param unit          the time unit of the {@code tickDuration}
         * @throws NullPointerException     if either of {@code threadFactory} and {@code unit} is {@code null}
         * @throws IllegalArgumentException if {@code tickDuration} is <= 0
         */
        public HashedWheelTimer(
                ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
        
            this(threadFactory, tickDuration, unit, 512);
        }
    
        /**
         * Creates a new timer.
         *
         * @param threadFactory a {@link ThreadFactory} that creates a
         *                      background {@link Thread} which is dedicated to
         *                      {@link TimerTask} execution.
         * @param tickDuration  the duration between tick
         * @param unit          the time unit of the {@code tickDuration}
         * @param ticksPerWheel the size of the wheel
         * @throws NullPointerException     if either of {@code threadFactory} and {@code unit} is {@code null}
         * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
         */
        public HashedWheelTimer(
                ThreadFactory threadFactory,
                long tickDuration, TimeUnit unit, int ticksPerWheel) {
        
            this(threadFactory, tickDuration, unit, ticksPerWheel, -1);
        }
    
        /**
         * Creates a new timer.
         * 初始化定时器 参数校验
         *
         * @param threadFactory      a {@link ThreadFactory} that creates a
         *                           background {@link Thread} which is dedicated to
         *                           {@link TimerTask} execution.
         * @param tickDuration       the duration between tick
         * @param unit               the time unit of the {@code tickDuration}
         * @param ticksPerWheel      the size of the wheel
         * @param maxPendingTimeouts The maximum number of pending timeouts after which call to
         *                           {@code newTimeout} will result in
         *                           {@link RejectedExecutionException}
         *                           being thrown. No maximum pending timeouts limit is assumed if
         *                           this value is 0 or negative.
         * @throws NullPointerException     if either of {@code threadFactory} and {@code unit} is {@code null}
         * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
         */
        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();
            }
        }
    
        @Override
        protected void finalize() throws Throwable {
        
            try {
        
                super.finalize();
            } finally {
        
                // This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown. If
                // we have not yet shutdown then we want to make sure we decrement the active instance count.
                if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
        
                    INSTANCE_COUNTER.decrementAndGet();
                }
            }
        }
    
        /**
         * 创建桶数组并初始化（环形队列）
         * @param ticksPerWheel
         * @return
         */
        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;
        }
    
        /**
         * Int 正整数归一化处理 -> 2^n
         * @param ticksPerWheel
         * @return
         */
        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;
        }
    
        /**
         * Starts the background thread explicitly.  The background thread will
         * start automatically on demand even if you did not call this method.
         *
         * 启动工作线程
         *
         * @throws IllegalStateException if this timer has been
         *                               {@linkplain #stop() stopped} already
         */
        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.
                }
            }
        }
    
        /**
         * 终止工作线程
         * @return
         */
        @Override
        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 {
        
                        workerThread.join(100);
                    } catch (InterruptedException ignored) {
        
                        interrupted = true;
                    }
                }
    
                if (interrupted) {
        
                    Thread.currentThread().interrupt();
                }
            } finally {
        
                INSTANCE_COUNTER.decrementAndGet();
            }
            return worker.unprocessedTimeouts();
        }
    
        @Override
        public boolean isStop() {
        
            return WORKER_STATE_SHUTDOWN == WORKER_STATE_UPDATER.get(this);
        }
    
        @Override
        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;
        }
    
        /**
         * Returns the number of pending timeouts of this {@link Timer}.
         */
        public long pendingTimeouts() {
        
            return pendingTimeouts.get();
        }
    
        private static void reportTooManyInstances() {
        
            String resourceType = ClassUtils.simpleClassName(HashedWheelTimer.class);
            logger.error("You are creating too many " + resourceType + " instances. " +
                    resourceType + " is a shared resource that must be reused across the JVM," +
                    "so that only a few instances are created.");
        }
    
        /**
         * 指针转动一格的流程
         */
        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();
    
                /**
                 * 遍历
                 * 1.移除取消的任务
                 * 2.插入新增的任务
                 * 3.获取当前桶 处理到期任务
                 * 4.开始下一步
                 * 5.移除所有桶内取消或过期的任务 将未取消任务加入unprocessedTimeouts 清理已取消任务
                 */
                do {
        
                    final long deadline = waitForNextTick();
                    if (deadline > 0) {
        
                        int idx = (int) (tick & mask);
                        processCancelledTasks();
                        HashedWheelBucket bucket =
                                wheel[idx];
                        transferTimeoutsToBuckets();
                        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();
            }
    
            /**
             * 将新任务添加到应桶内（格子的链表）
             * 1.默认每次转动最多产生 100000 待插入的任务
             * 2.总步数 = 过期时间 / 步长
             * 3.剩余圈数 = 总步数 / 圈长
             * 4.取步数和已走格数较大值，避免步数被已走的格子超过去
             * 5.格子数归一化后 &mask 即返回对应格子位置
             * 6.取出对应的桶 插入任务
             */
            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);
                        }
                    }
                }
            }
    
            /**
             * 根据步长计算下一步时间点，到时间点后则转动一格 [1 毫秒(ms) = 1000 微妙(us) = 1000000 纳秒]
             * 1.到期时长 = 步长 * ( 步数 + 1 )
             * 2.sleepTimeMs = 到期时长 - 当前时长
             * 3.当前时长 小于 到期时长 1纳秒 时      ===> sleepTimeMs = 1
             *   当前时长 大于 到期时长 999999纳秒 时 ===> sleepTimeMs = 0
             * 4.如果是Windows系统 除10取整 再乘10
             * 5.最终返回 当前时长
             */
            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);
            }
        }
    
        /**
         * 1、双向链表节点 包含TimerTask任务属性 任务提交后返回对应句柄
         */
        private static final class HashedWheelTimeout implements Timeout {
        
    
            /**
             * 节点（任务）状态 0-初始化 1-取消 2-到期
             */
            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");
    
            /**
             * 定时器 定时任务 到期时间
             */
            private final HashedWheelTimer timer;
            private final TimerTask task;
            private final long deadline;
    
            @SuppressWarnings({
        "unused", "FieldMayBeFinal", "RedundantFieldInitialization"})
            private volatile int state = ST_INIT;
    
            /**
             * 剩余圈数（当延时 任务 时长大于一个时间轮周期 则用圈数表示）
             */
            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;
            }
    
            /**
             * 1.自旋锁 保证操作原子性 只能基于初始化的状态去取消任务
             * 2.将取消的任务添加到取消任务集合
             */
            @Override
            public boolean cancel() {
        
                if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
        
                    return false;
                }
                // 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;
            }
    
            /**
             * 将任务从链表（桶）移除
             */
            void remove() {
        
                HashedWheelBucket bucket = this.bucket;
                if (bucket != null) {
        
                    bucket.remove(this);
                } else {
        
                    timer.pendingTimeouts.decrementAndGet();
                }
            }
    
            /**
             * 自旋锁 保证操作原子性 只能基于初始化的状态去取消任务
             * @param expected
             * @param state
             * @return
             */
            public boolean compareAndSetState(int expected, int state) {
        
                return STATE_UPDATER.compareAndSet(this, expected, state);
            }
    
            /**
             * 返回状态
             * @return
             */
            public int state() {
        
                return state;
            }
    
            @Override
            public boolean isCancelled() {
        
                return state() == ST_CANCELLED;
            }
    
            @Override
            public boolean isExpired() {
        
                return state() == ST_EXPIRED;
            }
    
            /**
             * 1.置为过期 操作失败则退出
             * 2.成功则执行任务
             */
            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() {
        
                final long currentTime = System.nanoTime();
                long remaining = deadline - currentTime + timer.startTime;
                String simpleClassName = ClassUtils.simpleClassName(this.getClass());
    
                StringBuilder buf = new StringBuilder(192)
                        .append(simpleClassName)
                        .append('(')
                        .append("deadline: ");
                if (remaining > 0) {
        
                    buf.append(remaining)
                            .append(" ns later");
                } else if (remaining < 0) {
        
                    buf.append(-remaining)
                            .append(" ns ago");
                } else {
        
                    buf.append("now");
                }
    
                if (isCancelled()) {
        
                    buf.append(", cancelled");
                }
    
                return buf.append(", task: ")
                        .append(task())
                        .append(')')
                        .toString();
            }
        }
    
        /**
         * 桶-时间轮的槽，即每个刻度格
         */
        private static final class HashedWheelBucket {
        
    
            /**
             * 头、尾
             */
            private HashedWheelTimeout head;
            private HashedWheelTimeout tail;
    
            /**
             * 向桶（链表）添加节点
             */
            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;
                }
            }
    
            /**
             * 遍历所有节点 若到期则移除并执行 若取消则直接移除 否则任务待执行周期-1
             */
            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;
                }
            }
    
            /**
             * 1.如果为中间节点，则移除当前节点
             * 2.如果当前节点为头，且为尾，则头、尾置为 空
             *   如果当前节点为头，不为尾，则下一节点置为 头
             *   如果当前节点为尾，则上一节点置为 尾
             * 3.上 下节点、桶置为 空
             * @param timeout
             * @return
             */
            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;
            }
    
            /**
             * 清理所有取消或过期的任务
             */
            void clearTimeouts(Set<Timeout> set) {
        
                for (; ; ) {
        
                    HashedWheelTimeout timeout = pollTimeout();
                    if (timeout == null) {
        
                        return;
                    }
                    if (timeout.isExpired() || timeout.isCancelled()) {
        
                        continue;
                    }
                    set.add(timeout);
                }
            }
    
            /**
             * 取出头节点
             * @return
             */
            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;
            }
        }
    
        /**
         * 判断是否为Windows系统
         */
        private static final boolean IS_OS_WINDOWS = System.getProperty("os.name", "").toLowerCase(Locale.US).contains("win");
        
        private boolean isWindows() {
        
            return IS_OS_WINDOWS;
        }
    }

### 2.任务处理接口类：Timeout ###

### 3.定时器接口类：Timer ###

### 4.任务接口类：TimerTask ###

## 三、调用： ##

### 1.测试接口： ###

package com.demo.controller;
    
    import com.demo.config.wheel.HashedWheelTimer;
    import com.demo.config.wheel.Timeout;
    import com.demo.config.wheel.TimerTask;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Description: 时间轮延迟
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-06 22:19
     * @version: V1.0.0
     */
    @RestController
    @RequestMapping("/delay")
    @Slf4j
    public class WheelController {
        
    
        private HashedWheelTimer wheelTimer = new HashedWheelTimer();
        private Timeout timeout;
    
        /**
         * 添加一个任务，并启动时间轮
         */
        public WheelController(){
        
            TimerTask task = timeout -> {
        
                log.info("时间轮启动，我是第一格：{}", TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()));
            };
            timeout = wheelTimer.newTimeout(task,1,TimeUnit.SECONDS);
            wheelTimer.start();
        }
    
        /**
         * 添加任务，延迟指定时间后执行
         * @param delay
         */
        @GetMapping("/wheel")
        public void wheel(int delay){
        
            long current = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis());
            TimerTask task = timeout1 -> {
        
                log.info("任务创建于：{} 延迟 [{}] 秒后执行，当前时间：{}",current,delay,TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()));
            };
            wheelTimer.newTimeout(task,delay,TimeUnit.SECONDS);
        }
    }

### 2.调试效果： ###

分别下发三个任务，查看触发效果：

顺序如下：
    1.http://127.0.0.1:8899/delay/wheel?delay=60
    2.http://127.0.0.1:8899/delay/wheel?delay=30
    3.http://127.0.0.1:8899/delay/wheel?delay=90

打印日志信息：

先下发延期60秒的任务，立即下发延期30秒的任务；则实际30秒任务会先到期，先执行

.   ____          _            __ _ _
     /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
    ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
     \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
      '  |____| .__|_| |_|_| |_\__, | / / / /
     =========|_|==============|___/=/_/_/_/
     :: Spring Boot ::                (v2.5.7)
    
    2022-01-06 22:57:49.746  INFO 12524 --- [           main] c.d.DelayApplication                     : Starting DelayApplication using Java 17.0.1 on zhx_yue with PID 12524 (E:\IdeaProjects\delay-demo\target\classes started by Administrator in E:\IdeaProjects\delay-demo)
    2022-01-06 22:57:49.750  INFO 12524 --- [           main] c.d.DelayApplication                     : No active profile set, falling back to default profiles: default
    2022-01-06 22:57:50.043  INFO 12524 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode!
    2022-01-06 22:57:50.044  INFO 12524 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data Redis repositories in DEFAULT mode.
    2022-01-06 22:57:50.056  INFO 12524 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 4 ms. Found 0 Redis repository interfaces.
    2022-01-06 22:57:50.283  INFO 12524 --- [           main] o.s.b.w.e.t.TomcatWebServer              : Tomcat initialized with port(s): 8899 (http)
    1月 06, 2022 10:57:50 下午 org.apache.coyote.AbstractProtocol init
    信息: Initializing ProtocolHandler ["http-nio-8899"]
    1月 06, 2022 10:57:50 下午 org.apache.catalina.core.StandardService startInternal
    信息: Starting service [Tomcat]
    1月 06, 2022 10:57:50 下午 org.apache.catalina.core.StandardEngine startInternal
    信息: Starting Servlet engine: [Apache Tomcat/9.0.55]
    1月 06, 2022 10:57:50 下午 org.apache.catalina.core.ApplicationContext log
    信息: Initializing Spring embedded WebApplicationContext
    2022-01-06 22:57:50.329  INFO 12524 --- [           main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 558 ms
    2022-01-06 22:57:50.652  INFO 12524 --- [           main] o.q.i.StdSchedulerFactory                : Using default implementation for ThreadExecutor
    2022-01-06 22:57:50.658  INFO 12524 --- [           main] o.q.c.SchedulerSignalerImpl              : Initialized Scheduler Signaller of type: class org.quartz.core.SchedulerSignalerImpl
    2022-01-06 22:57:50.658  INFO 12524 --- [           main] o.q.c.QuartzScheduler                    : Quartz Scheduler v.2.3.2 created.
    2022-01-06 22:57:50.658  INFO 12524 --- [           main] o.q.s.RAMJobStore                        : RAMJobStore initialized.
    2022-01-06 22:57:50.659  INFO 12524 --- [           main] o.q.c.QuartzScheduler                    : Scheduler meta-data: Quartz Scheduler (v2.3.2) 'quartzScheduler' with instanceId 'NON_CLUSTERED'
      Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.
      NOT STARTED.
      Currently in standby mode.
      Number of jobs executed: 0
      Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 10 threads.
      Using job-store 'org.quartz.simpl.RAMJobStore' - which does not support persistence. and is not clustered.
    
    2022-01-06 22:57:50.659  INFO 12524 --- [           main] o.q.i.StdSchedulerFactory                : Quartz scheduler 'quartzScheduler' initialized from an externally provided properties instance.
    2022-01-06 22:57:50.659  INFO 12524 --- [           main] o.q.i.StdSchedulerFactory                : Quartz scheduler version: 2.3.2
    2022-01-06 22:57:50.659  INFO 12524 --- [           main] o.q.c.QuartzScheduler                    : JobFactory set to: org.springframework.scheduling.quartz.SpringBeanJobFactory@7e191fda
    1月 06, 2022 10:57:50 下午 org.apache.coyote.AbstractProtocol start
    信息: Starting ProtocolHandler ["http-nio-8899"]
    2022-01-06 22:57:50.696  INFO 12524 --- [           main] o.s.b.w.e.t.TomcatWebServer              : Tomcat started on port(s): 8899 (http) with context path ''
    2022-01-06 22:57:51.021  INFO 12524 --- [           main] o.s.s.q.SchedulerFactoryBean             : Starting Quartz Scheduler now
    2022-01-06 22:57:51.021  INFO 12524 --- [           main] o.q.c.QuartzScheduler                    : Scheduler quartzScheduler_$_NON_CLUSTERED started.
    2022-01-06 22:57:51.040  INFO 12524 --- [           main] c.d.DelayApplication                     : Started DelayApplication in 1.526 seconds (JVM running for 2.057)
    2022-01-06 22:57:51.564  INFO 12524 --- [pool-2-thread-1] c.d.c.WheelController                    : 时间轮启动，我是第一格：1641481071
    1月 06, 2022 10:58:06 下午 org.apache.catalina.core.ApplicationContext log
    信息: Initializing Spring DispatcherServlet 'dispatcherServlet'
    2022-01-06 22:58:06.496  INFO 12524 --- [nio-8899-exec-1] o.s.w.s.DispatcherServlet                : Initializing Servlet 'dispatcherServlet'
    2022-01-06 22:58:06.496  INFO 12524 --- [nio-8899-exec-1] o.s.w.s.DispatcherServlet                : Completed initialization in 0 ms
    2022-01-06 22:58:44.064  INFO 12524 --- [pool-2-thread-1] c.d.c.WheelController                    : 任务创建于：1641481093 延迟 [30] 秒后执行，当前时间：1641481124
    2022-01-06 22:59:06.564  INFO 12524 --- [pool-2-thread-1] c.d.c.WheelController                    : 任务创建于：1641481086 延迟 [60] 秒后执行，当前时间：1641481146
    2022-01-06 22:59:51.063  INFO 12524 --- [pool-2-thread-1] c.d.c.WheelController                    : 任务创建于：1641481100 延迟 [90] 秒后执行，当前时间：1641481191

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/88d40977e7774ccea9fd4f89723edd9e.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_18_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/ad1f377e27864d81b28fdfd393b86b39.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_18,color_FFFFFF,t_70,g_se,x_16