详解时间轮TimeWheel的工作原理

public interface Timer {

    /**
     * 启动时间轮
     * */
    void startTimeWheel();

    /**
     * 创建新的超时任务(必须先startTimeWheel完成后，才能创建新任务)
     * @param task 超时时需要调度的自定义任务
     * @param delayTime 延迟时间
     * @param timeUnit 延迟时间delayTime的单位
     * */
    void newTimeoutTask(Runnable task, long delayTime, TimeUnit timeUnit);
}

/**
 * 参考netty实现的单层时间轮
 * */
public class MyHashedTimeWheel implements Timer{

    /**
     * 环形数组
     * */
    private final MyHashedTimeWheelBucket[] ringBucketArray;

    /**
     * 世间轮启动时的具体时间戳(单位：纳秒nanos)
     * */
    private long startTime;

    /**
     * 是否已启动
     * */
    private final AtomicBoolean started = new AtomicBoolean(false);

    /**
     * 时间轮每次转动的时间(单位：纳秒nanos)
     * (perTickTime越短，调度会更精确，但cpu开销也会越大)
     * */
    private final long perTickTime;

    /**
     * 总tick数
     * */
    private long totalTick = 0;

    /**
     * 待处理任务的队列
     * (多外部生产者写入，时间轮内的单worker消费者读取，所以netty的实现里使用了效率更高的MpscQueue，Mpsc即MultiProducerSingleConsumer)
     * */
    private final Queue<MyTimeoutTaskNode> unProcessTaskQueue = new LinkedBlockingDeque<>();

    /**
     * 用于实际执行到期任务的线程池
     * */
    private final Executor taskExecutor;

    private Thread workerThread;

    /**
     * 构造函数
     * */
    public MyHashedTimeWheel(int ringArraySize, long perTickTime, Executor taskExecutor) {
        this.ringBucketArray = new MyHashedTimeWheelBucket[ringArraySize];
        for(int i=0; i<ringArraySize; i++){
            // 初始化，填充满时间轮唤醒数组
            this.ringBucketArray[i] = new MyHashedTimeWheelBucket();
        }

        this.perTickTime = perTickTime;
        this.taskExecutor = taskExecutor;
    }

    /**
     * 启动worker线程等初始化操作，必须执行完成后才能正常工作
     * (简单起见，和netty不一样不是等任务被创建时才懒加载的，必须提前启动)
     * */
    @Override
    public void startTimeWheel(){
        // 启动worker线程
        this.workerThread = new Thread(new Worker());
        this.workerThread.start();

        while (!this.started.get()){
            // 自旋循环，等待一会
        }

        System.out.println("startTimeWheel 启动完成:" + this.getClass().getSimpleName());
    }

    @Override
    public void newTimeoutTask(Runnable task, long delayTime, TimeUnit timeUnit){
        long deadline = System.nanoTime() + timeUnit.toNanos(delayTime);

        // Guard against overflow.
        if (delayTime > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }

        MyTimeoutTaskNode newTimeoutTaskNode = new MyTimeoutTaskNode();
        newTimeoutTaskNode.setTargetTask(task);
        newTimeoutTaskNode.setDeadline(deadline);

        unProcessTaskQueue.add(newTimeoutTaskNode);
    }

    private final class Worker implements Runnable{

        @Override
        public void run() {
            MyHashedTimeWheel.this.startTime = System.nanoTime();

            // 启动
            MyHashedTimeWheel.this.started.set(true);

            // 简单起见，不考虑优雅启动和暂停的逻辑
            while (true){
                // 等待perTick
                waitForNextTick();

                // 在捞取当前tick下需要处理的bucket前，先将加入到队列中的任务转移到环形数组中(可能包含在当前tick下就要处理的任务)
                transferTaskToBuckets();

                // 基于总tick数，对环形数组的长度取模，计算出当前tick下需要处理的bucket桶的下标
                int idx = (int) (MyHashedTimeWheel.this.totalTick % MyHashedTimeWheel.this.ringBucketArray.length);
                MyHashedTimeWheelBucket bucket = MyHashedTimeWheel.this.ringBucketArray[idx];
                // 处理当前插槽内的任务(遍历链表中的所有任务，round全部减一，如果减为负数了则说明这个任务超时到期了，将其从链表中移除后并交给线程池执行指定的任务)
                bucket.expireTimeoutTask(MyHashedTimeWheel.this.taskExecutor);
                // 循环tick一次，总tick数自增1
                MyHashedTimeWheel.this.totalTick++;
            }
        }

        /**
         * per tick时钟跳动，基于Thread.sleep
         * */
        private void waitForNextTick(){
            // 由于Thread.sleep并不是绝对精确的被唤醒，所以只能通过(('总的tick数+1' * '每次tick的间隔') + '时间轮启动时间')来计算精确的下一次tick时间
            // 而不能简单的Thread.sleep(每次tick的间隔)

            long nextTickTime = (MyHashedTimeWheel.this.totalTick + 1) * MyHashedTimeWheel.this.perTickTime
                            + MyHashedTimeWheel.this.startTime;

            // 因为nextTickTime是纳秒，sleep需要的是毫秒，需要保证纳秒数过小时，导致直接计算出来的毫秒数为0
            // 因此(‘实际休眠的纳秒数'+999999)/1000000,保证了纳秒转毫秒时，至少会是1毫秒，而不会出现sleep(0毫秒)令cpu空转
            long needSleepTime = (nextTickTime - System.nanoTime() + 999999) / 1000000;
            try {
                // 比起netty，忽略了一些处理特殊场景bug的逻辑
                Thread.sleep(needSleepTime);
            } catch (InterruptedException ignored) {

            }
        }

        private void transferTaskToBuckets() {
            // 为了避免worker线程在一次循环中处理太多的任务，所以直接限制了一个最大值100000
            // 如果真的有这么多，就等到下次tick循环的时候再去做。
            // 因为这个操作是cpu密集型的，处理太多的话，可能导致无法在一个短的tick周期内完成一次循环
            for (int i = 0; i < 100000; i++) {
                MyTimeoutTaskNode timeoutTaskNode = MyHashedTimeWheel.this.unProcessTaskQueue.poll();
                if (timeoutTaskNode == null) {
                    // 队列为空了，直接结束
                    return;
                }

                // 计算到任务超时时，应该执行多少次tick
                // (和netty里的不一样，这里的deadline是超时时间的绝对时间，所以需要先减去时间轮的startTime)
                // (netty中是生产者线程在add时事先减去了startTime，比起由worker线程统一处理效率更高，但个人觉得这里的写法会更直观)
                long totalTickWhenTimeout = (timeoutTaskNode.getDeadline() - MyHashedTimeWheel.this.startTime) / MyHashedTimeWheel.this.perTickTime;
                // 减去当前时间轮已经进行过的tick数量
                long remainingTickWhenTimeout = (totalTickWhenTimeout - MyHashedTimeWheel.this.totalTick);
                // 因为一次时间轮旋转会经过ringBucketArray.length次tick，所以求个余数
                long remainingRounds = remainingTickWhenTimeout / MyHashedTimeWheel.this.ringBucketArray.length;
                // 计算出当前任务需要转多少圈之后才会超时
                timeoutTaskNode.setRounds(remainingRounds);

                // 如果传入的deadline早于当前系统时间，则totalTickWhenTimeout可能会小于当前的totalTick
                // 这种情况下，让这个任务在当前tick下就立即超时而被调度是最合理的，而不能在求余后放到一个错误的位置而等一段时间才调度（所以必须取两者的最大值）
                final long ticks = Math.max(totalTickWhenTimeout, MyHashedTimeWheel.this.totalTick); // Ensure we don't schedule for past.
                // 如果能限制环形数组的长度为2的幂，则可以改为ticks & mask，位运算效率更高
                int stopIndex = (int) (ticks % MyHashedTimeWheel.this.ringBucketArray.length);
                MyHashedTimeWheelBucket bucket = MyHashedTimeWheel.this.ringBucketArray[stopIndex];
                // 计算并找到应该被放置的那个bucket后，将其插入当前bucket指向的链表中
                bucket.addTimeout(timeoutTaskNode);
            }
        }
    }
}

/**
 * 时间轮环形数组下标对应的桶(保存一个超时任务MyTimeoutTaskNode的链表)
 * */
public class MyHashedTimeWheelBucket {

    private final LinkedList<MyTimeoutTaskNode> linkedList = new LinkedList<>();

    public void addTimeout(MyTimeoutTaskNode timeout) {
        linkedList.add(timeout);
    }

    /**
     * 遍历链表中的所有任务，round全部减一，如果减为负数了则说明这个任务超时到期了，将其从链表中移除后并交给线程池执行指定的任务
     * */
    public void expireTimeoutTask(Executor executor){
        Iterator<MyTimeoutTaskNode> iterator = linkedList.iterator();
        while(iterator.hasNext()){
            MyTimeoutTaskNode currentNode = iterator.next();
            long currentNodeRound = currentNode.getRounds();
            if(currentNodeRound <= 0){
                // 将其从链表中移除
                iterator.remove();
                // count小于等于0，说明超时了，交给线程池去异步执行
                executor.execute(currentNode.getTargetTask());
            }else{
                // 当前节点还未超时，round自减1
                currentNode.setRounds(currentNodeRound-1);
            }

            // 简单起见，不考虑任务被外部自己取消的case(netty里的timeout.isCancelled())
        }
    }
}

public class MyTimeoutTaskNode {

    /**
     * 任务具体的到期时间(绝对时间)
     * */
    private long deadline;

    /**
     * 存储在时间轮中，需要等待的轮次
     * (rounds在初始化后，每次时间轮转动一周便自减1，当减为0时便代表当前任务需要被调度)
     * */
    private long rounds;

    /**
     * 创建任务时，用户指定的到期时进行调度的任务
     * */
    private Runnable targetTask;

    public long getDeadline() {
        return deadline;
    }

    public void setDeadline(long deadline) {
        this.deadline = deadline;
    }

    public long getRounds() {
        return rounds;
    }

    public void setRounds(long rounds) {
        this.rounds = rounds;
    }

    public Runnable getTargetTask() {
        return targetTask;
    }

    public void setTargetTask(Runnable targetTask) {
        this.targetTask = targetTask;
    }
}

/**
 * 层次时间轮，会存在空转问题
 * */
public class MyHierarchicalHashedTimerV1 implements Timer {

    /**
     * 是否已启动
     * */
    private AtomicBoolean started = new AtomicBoolean(false);

    /**
     * 世间轮启动时的具体时间戳(单位：纳秒nanos)
     * */
    private long startTime;

    /**
     * 时间轮每次转动的时间(单位：纳秒nanos)
     * (perTickTime越短，调度会更精确，但cpu开销也会越大)
     * */
    private final long perTickTime;

    /**
     * 总tick数
     * */
    private long totalTick = 0;

    /**
     * 待处理任务的队列
     * (多外部生产者写入，时间轮内的单worker消费者读取，所以netty的实现里使用了效率更高的MpscQueue，Mpsc即MultiProducerSingleConsumer)
     * */
    private final Queue<MyTimeoutTaskNode> unProcessTaskQueue = new LinkedBlockingDeque<>();

    /**
     * timer持有的最低层的时间轮
     * */
    private final MyHierarchicalHashedTimeWheelV1 lowestTimeWheel;

    /**
     * 构造函数
     * */
    public MyHierarchicalHashedTimerV1(int ringArraySize, long perTickTime, Executor taskExecutor) {
        this.perTickTime = perTickTime;

        // 初始化最底层的时间轮
        this.lowestTimeWheel = new MyHierarchicalHashedTimeWheelV1(ringArraySize,perTickTime,taskExecutor,0);
    }

    /**
     * 启动worker线程等初始化操作，必须执行完成后才能正常工作
     * (简单起见，和netty不一样不是等任务被创建时才懒加载的，必须提前启动)
     * */
    @Override
    public void startTimeWheel(){
        // 启动worker线程
        new Thread(new Worker()).start();

        while (!this.started.get()){
            // 自旋循环，等待一会
        }

        System.out.println("startTimeWheel 启动完成:" + this.getClass().getSimpleName());
    }

    @Override
    public void newTimeoutTask(Runnable task, long delayTime, TimeUnit timeUnit){
        long deadline = System.nanoTime() + timeUnit.toNanos(delayTime);

        // Guard against overflow.
        if (delayTime > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }

        MyTimeoutTaskNode newTimeoutTaskNode = new MyTimeoutTaskNode();
        newTimeoutTaskNode.setTargetTask(task);
        newTimeoutTaskNode.setDeadline(deadline);

        this.unProcessTaskQueue.add(newTimeoutTaskNode);
    }

    private final class Worker implements Runnable{

        @Override
        public void run() {
            MyHierarchicalHashedTimerV1.this.startTime = System.nanoTime();
            // 启动
            MyHierarchicalHashedTimerV1.this.started.set(true);

            // 简单起见，不考虑优雅启动和暂停的逻辑
            while (true){
                // 等待perTick
                waitForNextTick();

                // 在捞取当前tick下需要处理的bucket前，先将加入到队列中的任务转移到时间轮中(可能包含在当前tick下就要处理的任务)
                // 层级时间轮内部会做进一步的分配(放不下的话就溢出到更上一层的时间轮)
                transferTaskToTimeWheel();

                // 推进时间轮(层级时间轮内部满了一圈就会进一步的推进更上一层的时间轮)
                MyHierarchicalHashedTimerV1.this.lowestTimeWheel.advanceClockByTick(
                    (taskNode)->
                        // 参考kafka的写法，避免Timer里的一些属性被传到各个bucket里面
                        MyHierarchicalHashedTimerV1.this.lowestTimeWheel
                            .addTimeoutTask(MyHierarchicalHashedTimerV1.this.startTime, taskNode)
                );

                // 循环tick一次，总tick数自增1
                MyHierarchicalHashedTimerV1.this.totalTick++;
            }
        }

        /**
         * per tick时钟跳动，基于Thread.sleep
         * */
        private void waitForNextTick(){
            // 由于Thread.sleep并不是绝对精确的被唤醒，所以只能通过(('总的tick数+1' * '每次tick的间隔') + '时间轮启动时间')来计算精确的下一次tick时间
            // 而不能简单的Thread.sleep(每次tick的间隔)

            long nextTickTime = (MyHierarchicalHashedTimerV1.this.totalTick + 1) * MyHierarchicalHashedTimerV1.this.perTickTime
                + MyHierarchicalHashedTimerV1.this.startTime;

            // 因为nextTickTime是纳秒，sleep需要的是毫秒，需要保证纳秒数过小时，导致直接计算出来的毫秒数为0
            // 因此(‘实际休眠的纳秒数'+999999)/1000000,保证了纳秒转毫秒时，至少会是1毫秒，而不会出现sleep(0毫秒)令cpu空转
            long needSleepTime = (nextTickTime - System.nanoTime() + 999999) / 1000000;
            
            try {
                // 比起netty，忽略了一些处理特殊场景bug的逻辑
                Thread.sleep(needSleepTime);
            } catch (InterruptedException ignored) {

            }
        }

        /**
         * 加入到队列中的任务转移到时间轮中
         * */
        private void transferTaskToTimeWheel() {
            // 为了避免worker线程在一次循环中处理太多的任务，所以直接限制了一个最大值100000
            // 如果真的有这么多，就等到下次tick循环的时候再去做。
            // 因为这个操作是cpu密集型的，处理太多的话，可能导致无法在一个短的tick周期内完成一次循环
            for (int i = 0; i < 100000; i++) {
                MyTimeoutTaskNode timeoutTaskNode = MyHierarchicalHashedTimerV1.this.unProcessTaskQueue.poll();
                if (timeoutTaskNode == null) {
                    // 队列为空了，直接结束
                    return;
                }

                // 层级时间轮内部会做进一步的分配(放不下的话就溢出到更上一层的时间轮)
                MyHierarchicalHashedTimerV1.this.lowestTimeWheel.addTimeoutTask(
                    MyHierarchicalHashedTimerV1.this.startTime, timeoutTaskNode);
            }
        }
    }
}

public class MyHierarchicalHashedTimeWheelV1 {

    private final MyHierarchyHashedTimeWheelBucketV1[] ringBucketArray;

    /**
     * 总tick数
     * */
    private long totalTick = 0;

    /**
     * 当前时间轮所能承载的时间间隔
     * */
    private final long interval;

    /**
     * 时间轮每次转动的时间(单位：纳秒nanos)
     * (perTickTime越短，调度会更精确，但cpu开销也会越大)
     * */
    private final long perTickTime;

    /**
     * 上一层时间跨度更大的时间轮
     * */
    private MyHierarchicalHashedTimeWheelV1 overFlowWheel;

    /**
     * 用于实际执行到期任务的线程池
     * */
    private final Executor taskExecutor;

    /**
     * 是否是最底层的时间轮（只有最底层的时间轮才真正的对任务进行调度）
     * */
    private final int level;

    public MyHierarchicalHashedTimeWheelV1(int ringArraySize,long perTickTime, Executor taskExecutor,int level) {
        this.ringBucketArray = new MyHierarchyHashedTimeWheelBucketV1[ringArraySize];
        for(int i=0; i<ringArraySize; i++){
            // 初始化，填充满时间轮唤醒数组
            this.ringBucketArray[i] = new MyHierarchyHashedTimeWheelBucketV1();
        }

        this.perTickTime = perTickTime;
        this.taskExecutor = taskExecutor;
        this.interval = perTickTime * ringArraySize;
        this.level = level;

        if(level > 0){
            this.totalTick = 1;
        }
    }

    /**
     * 当前时间轮加入任务(溢出的话，则需要放到上一层的时间轮中)
     * */
    public void addTimeoutTask(long startTime, MyTimeoutTaskNode timeoutTaskNode){
        long deadline = timeoutTaskNode.getDeadline();

        // 当前时间轮所能承载的最大绝对时间为：每个tick的间隔 * 插槽数 + (基于startTime的当前绝对时间)
        long currentWheelMaxRange = this.interval + (startTime + this.perTickTime * this.totalTick);

        if(deadline < currentWheelMaxRange){
            // 当前时间轮能够承载这个任务，无需放到上一层时间轮中

            // 计算到任务超时时，应该执行多少次tick
            // (和netty里的不一样，这里的deadline是超时时间的绝对时间，所以需要先减去时间轮的startTime)
            // (netty中是生产者线程在add时事先减去了startTime，比起由worker线程统一处理效率更高，但个人觉得这里的写法会更直观)
            long totalTickWhenTimeout = (deadline - startTime) / this.perTickTime;

            // 如果传入的deadline早于当前系统时间，则totalTickWhenTimeout可能会小于当前的totalTick
            // 这种情况下，让这个任务在当前tick下就立即超时而被调度是最合理的，而不能在求余后放到一个错误的位置而等一段时间才调度（所以必须取两者的最大值）
            final long ticks = Math.max(totalTickWhenTimeout, this.totalTick); // Ensure we don't schedule for past.
            // 如果能限制环形数组的长度为2的幂，则可以改为ticks & mask，位运算效率更高
            int stopIndex = (int) (ticks % this.ringBucketArray.length);

            MyHierarchyHashedTimeWheelBucketV1 bucket = this.ringBucketArray[stopIndex];
            // 计算并找到应该被放置的那个bucket后，将其插入当前bucket指向的链表中
            bucket.addTimeout(timeoutTaskNode);
        }else{
            // 当前时间轮无法承载这个任务，需要放到上一层时间轮中

            // 上层时间轮不存在，创建之
            if(this.overFlowWheel == null){
                // 上层时间轮的环形数组大小保持不变，perTick是当前时间轮的整个间隔(类似低层的60秒等于上一层的1分钟)
                this.overFlowWheel = new MyHierarchicalHashedTimeWheelV1(
                    this.ringBucketArray.length, this.interval, taskExecutor,this.level+1);
            }

            // 加入到上一层的时间轮中(对于较大的deadline，addTimeoutTask操作可能会递归数次，放到第N层的时间轮中)
            this.overFlowWheel.addTimeoutTask(startTime,timeoutTaskNode);
        }
    }

    public void advanceClockByTick(Consumer<MyTimeoutTaskNode> flushInLowerWheelFn){
        // 基于总tick数，对环形数组的长度取模，计算出当前tick下需要处理的bucket桶的下标
        int idx = (int) (this.totalTick % this.ringBucketArray.length);

        MyHierarchyHashedTimeWheelBucketV1 bucket = this.ringBucketArray[idx];

        if(this.level == 0){
            // 如果是最底层的时间轮，将当前tick下命中的bucket中的任务丢到taskExecutor中执行
            bucket.expireTimeoutTask(this.taskExecutor);
        }else{
            // 如果不是最底层的时间轮，将当前tick下命中的bucket中的任务交给下一层的时间轮
            // 这里转交到下一层有两种方式：第一种是从上到下的转交，另一种是当做新任务一样还是从最下层的时间轮开始放，放不下再往上溢出
            // 选用后一种逻辑，最大的复用已有的创建新任务的逻辑，会好理解一点
            bucket.flush(flushInLowerWheelFn);
        }

        // 当前时间轮的总tick自增1
        this.totalTick++;

        // 当前时间轮的总tick数满了一圈之后，推进上一层时间轮进行一次tick(如果上一层时间轮存在的话)
        if(this.totalTick % this.ringBucketArray.length == 0 && this.overFlowWheel != null){
            this.overFlowWheel.advanceClockByTick(flushInLowerWheelFn);
        }
    }
}

/**
 * 时间轮环形数组下标对应的桶(保存一个超时任务MyTimeoutTaskNode的链表)
 * */
public class MyHierarchyHashedTimeWheelBucketV1 {

    private final LinkedList<MyTimeoutTaskNode> linkedList = new LinkedList<>();

    public void addTimeout(MyTimeoutTaskNode timeout) {
        linkedList.add(timeout);
    }

    /**
     * 遍历链表中的所有任务，round全部减一，如果减为负数了则说明这个任务超时到期了，将其从链表中移除后并交给线程池执行指定的任务
     * */
    public void expireTimeoutTask(Executor executor){
        Iterator<MyTimeoutTaskNode> iterator = linkedList.iterator();
        while(iterator.hasNext()){
            MyTimeoutTaskNode currentNode = iterator.next();
            long currentNodeRound = currentNode.getRounds();
            if(currentNodeRound <= 0){
                // 将其从链表中移除
                iterator.remove();
                // count小于等于0，说明超时了，交给线程池去异步执行
                executor.execute(currentNode.getTargetTask());
            }else{
                // 当前节点还未超时，round自减1
                currentNode.setRounds(currentNodeRound-1);
            }

            // 简单起见，不考虑任务被外部自己取消的case(netty里的timeout.isCancelled())
        }
    }

    /**
     * 将当前bucket中的数据，通过flushInLowerWheelFn，全部转移到更底层的时间轮中
     * */
    public void flush(Consumer<MyTimeoutTaskNode> flushInLowerWheelFn){
        Iterator<MyTimeoutTaskNode> iterator = linkedList.iterator();
        while(iterator.hasNext()){
            MyTimeoutTaskNode currentNode = iterator.next();
            // 先从链表中移除
            iterator.remove();
            // 通过flushInLowerWheelFn，转移到更底层的时间轮中
            flushInLowerWheelFn.accept(currentNode);

            // 简单起见，不考虑任务被外部自己取消的case(netty里的timeout.isCancelled())
        }
    }
}

public class MyHierarchicalHashedTimerV2 implements Timer {

    /**
     * 是否已启动
     * */
    private AtomicBoolean started = new AtomicBoolean(false);

    /**
     * 关联的最底层时间轮
     * */
    private volatile MyHierarchicalHashedTimeWheelV2 lowestTimeWheel;

    /**
     * 时间轮的启动时间（单位:纳秒）
     * */
    private long startTime;

    /**
     * 每次tick的间隔（单位:纳秒）
     * */
    private final long perTickTime;

    /**
     * 时间轮的大小
     * */
    private final int timeWheelSize;

    /**
     * 用于实际执行到期任务的线程池
     * */
    private final Executor taskExecutor;

    /**
     * 用于存储bucket元素的延迟队列，用于解决时间轮空转的问题
     * */
    private final DelayQueue<MyHierarchyHashedTimeWheelBucketV2> bucketDelayQueue = new DelayQueue<>();

    public MyHierarchicalHashedTimerV2(int timeWheelSize,long perTickTime, Executor taskExecutor) {
        this.timeWheelSize = timeWheelSize;
        this.perTickTime = perTickTime;
        this.taskExecutor = taskExecutor;
    }

    /**
     * 启动worker线程等初始化操作，必须执行完成后才能正常工作
     * (简单起见，和netty不一样不是等任务被创建时才懒加载的，必须提前启动)
     * */
    @Override
    public void startTimeWheel(){
        // 启动worker线程
        new Thread(new Worker()).start();

        while (!this.started.get()){
            // 自旋循环，等待一会
        }

        System.out.println("startTimeWheel 启动完成:" + this.getClass().getSimpleName());
    }

    @Override
    public void newTimeoutTask(Runnable task, long delayTime, TimeUnit timeUnit){
        long deadline = System.nanoTime() + timeUnit.toNanos(delayTime);

        // Guard against overflow.
        if (delayTime > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }

        MyTimeoutTaskNode newTimeoutTaskNode = new MyTimeoutTaskNode();
        newTimeoutTaskNode.setTargetTask(task);
        newTimeoutTaskNode.setDeadline(deadline);

        // 加入到最底层的时间轮中，当前时间轮放不下的会溢出都上一层时间轮
        this.lowestTimeWheel.addTimeoutTask(newTimeoutTaskNode);
    }

    private void advanceClock(){
        try {
            MyHierarchyHashedTimeWheelBucketV2 bucket = this.bucketDelayQueue.take();
            lowestTimeWheel.advanceClockByTick(bucket.getExpiration());
            bucket.flush((node)->{
                // 当前选中的bucket中的任务，重新插入到时间轮中
                // 1 原本处于高层的bucket中的任务会被放到更底层
                // 2 原本就处于最低一层的bucket中的任务会被直接执行
                this.lowestTimeWheel.addTimeoutTask(node);
            });
            // 将当前时间轮的数据
        } catch (Exception e) {
            // 忽略掉异常
            e.printStackTrace();
        }
    }


    private final class Worker implements Runnable {
        @Override
        public void run() {
            MyHierarchicalHashedTimerV2.this.startTime = System.nanoTime();

            // 初始化最底层的时间轮
            MyHierarchicalHashedTimerV2.this.lowestTimeWheel = new MyHierarchicalHashedTimeWheelV2(
                MyHierarchicalHashedTimerV2.this.startTime,
                MyHierarchicalHashedTimerV2.this.perTickTime,
                MyHierarchicalHashedTimerV2.this.timeWheelSize,
                MyHierarchicalHashedTimerV2.this.taskExecutor,
                MyHierarchicalHashedTimerV2.this.bucketDelayQueue
            );

            // 启动
            MyHierarchicalHashedTimerV2.this.started.set(true);

            while (true){
                // 一直无限循环，不断推进时间
                advanceClock();
            }
        }
    }
}

public class MyHierarchicalHashedTimeWheelV2 {

    /**
     * 上层时间轮（生产者/消费者都会访问到，volatile修饰）
     * */
    private volatile MyHierarchicalHashedTimeWheelV2 overflowTimeWheel;

    /**
     * 每次tick的间隔（单位:纳秒）
     * */
    private final long perTickTime;

    /**
     * 时间轮环形数组
     * */
    private final MyHierarchyHashedTimeWheelBucketV2[] ringBucketArray;

    /**
     * 用于实际执行到期任务的线程池
     * */
    private final Executor taskExecutor;

    /**
     * 时间轮的当前时间
     * */
    private long currentTime;

    /**
     * 当前时间轮的间隔(每次tick的时间 * 时间轮的大小)
     * */
    private final long interval;

    private final DelayQueue<MyHierarchyHashedTimeWheelBucketV2> bucketDelayQueue;

    public MyHierarchicalHashedTimeWheelV2(long startTime, long perTickTime, int wheelSize, Executor taskExecutor,
                                           DelayQueue<MyHierarchyHashedTimeWheelBucketV2> bucketDelayQueue) {
        // 初始化环形数组
        this.ringBucketArray = new MyHierarchyHashedTimeWheelBucketV2[wheelSize];
        for(int i=0; i<wheelSize; i++){
            this.ringBucketArray[i] = new MyHierarchyHashedTimeWheelBucketV2();
        }

        // 初始化时，当前时间为startTime
        this.currentTime = startTime - (startTime % perTickTime);
        this.perTickTime = perTickTime;
        this.taskExecutor = taskExecutor;
        this.interval = perTickTime * wheelSize;
        this.bucketDelayQueue = bucketDelayQueue;
    }

    public void addTimeoutTask(MyTimeoutTaskNode timeoutTaskNode) {
        long deadline = timeoutTaskNode.getDeadline();
        if(deadline < this.currentTime + this.perTickTime){
            // 超时时间小于1tick，直接执行
            this.taskExecutor.execute(timeoutTaskNode.getTargetTask());
        }else if(deadline < this.currentTime + this.interval){
            // 当前时间轮放的下

            // 在超时时，理论上总共需要的tick数
            long totalTick = deadline / this.perTickTime;

            // 如果传入的deadline早于当前系统时间，则totalTickWhenTimeout可能会小于当前的totalTick
            // 这种情况下，让这个任务在当前tick下就立即超时而被调度是最合理的，而不能在求余后放到一个错误的位置而等一段时间才调度（所以必须取两者的最大值）
            // 如果能限制环形数组的长度为2的幂，则可以改为ticks & mask，位运算效率更高
            int stopIndex = (int) (totalTick % this.ringBucketArray.length);

            MyHierarchyHashedTimeWheelBucketV2 bucket = this.ringBucketArray[stopIndex];
            // 计算并找到应该被放置的那个bucket后，将其插入当前bucket指向的链表中
            bucket.addTimeout(timeoutTaskNode);

            // deadline先除以this.perTickTime再乘以this.perTickTime,可以保证放在同一个插槽下的任务，expiration都是一样的
            long expiration = totalTick * this.perTickTime;
            boolean isNewRound = bucket.setExpiration(expiration);
            if(isNewRound){
                this.bucketDelayQueue.offer(bucket);
            }
        }else{
            // 当前时间轮放不下
            if(this.overflowTimeWheel == null){
                createOverflowWheel();
            }

            // 加入到上层的时间轮中(较大的deadline会递归多次)
            this.overflowTimeWheel.addTimeoutTask(timeoutTaskNode);
        }
    }

    /**
     * 推进当前时间轮的时钟
     * 举个例子：假设当前时间轮的当前时间是第10分钟，perTickTime是1分钟，
     * 1.如果expiration是第10分钟第1秒，则不用推动当前时间
     * 2.如果expiration是第11分钟第0秒，则需要推动当前时间
     * */
    public void advanceClockByTick(long expiration){
        // 只会在tick推进时才会被调用，参数expiration可以认为是当前时间轮的系统时间
        if(expiration >= this.currentTime + this.perTickTime){
            // 超过了1tick，则需要推进当前时间轮 (始终保持当前时间是perTickTime的整数倍，逻辑上的totalTick)
            this.currentTime = expiration - (expiration % this.perTickTime);
            if(this.overflowTimeWheel != null){
                // 如果上层时间轮存在，则递归的继续推进
                this.overflowTimeWheel.advanceClockByTick(expiration);
            }
        }
    }

    private synchronized void createOverflowWheel(){
        if(this.overflowTimeWheel == null){
            // 创建上层时间轮，上层时间轮的perTickTime = 当前时间轮的interval
            this.overflowTimeWheel = new MyHierarchicalHashedTimeWheelV2(
                this.currentTime, this.interval, this.ringBucketArray.length, this.taskExecutor, this.bucketDelayQueue);
        }
    }
}

public class MyHierarchyHashedTimeWheelBucketV2 implements Delayed {

    private final LinkedList<MyTimeoutTaskNode> taskList = new LinkedList<>();

    private final AtomicLong expiration = new AtomicLong(-1);

    public synchronized void addTimeout(MyTimeoutTaskNode timeout) {
        taskList.add(timeout);
    }

    public synchronized void flush(Consumer<MyTimeoutTaskNode> flush) {
        Iterator<MyTimeoutTaskNode> iterator = taskList.iterator();
        while (iterator.hasNext()){
            MyTimeoutTaskNode node = iterator.next();
            // 从当前bucket中移除，转移到更下层的时间轮中
            iterator.remove();
            flush.accept(node);

            // 简单起见，不考虑任务被外部自己取消的case(netty里的timeout.isCancelled())
        }

        this.expiration.set(-1L);
    }

    /**
     * 设置当前bucket的超时时间
     * @return 是否是一个新的bucket  true：是
     * */
    public boolean setExpiration(long expiration){
        long oldValue = this.expiration.getAndSet(expiration);

        // 如果不一样，说明当前的expiration已经超过了原来的expiration一圈了，逻辑上不再是同一个bucket
        return oldValue != expiration;
    }

    public long getExpiration(){
        return this.expiration.get();
    }

    @Override
    public long getDelay(TimeUnit unit) {
        // 还剩余多少时间过期
        long delayNanos = Math.max(this.expiration.get() - System.nanoTime(), 0);

        // 将纳秒单位基于unit转换
        return unit.convert(delayNanos,TimeUnit.NANOSECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if(o instanceof MyHierarchyHashedTimeWheelBucketV2){
            return Long.compare(this.expiration.get(),((MyHierarchyHashedTimeWheelBucketV2) o).expiration.get());
        }

        return 0;
    }
}