netty源码学习之-HashedWheelTimer

news2024/11/27 15:13:14

netty源码学习之-HashedWheelTimer

  • 概述
  • 使用
  • 相关概念解析
    • 时间轮 运行时序图
  • 源码
    • worker
    • HashedWheelTimeout
    • HashedWheelBucket

概述

该部分源码是netty的时间轮,netty的时间轮是单轮,其他时间轮是多轮设计,今天先了解下netty的时间轮设计

在这里插入图片描述

使用

public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 16);
        System.out.println("current timestamp="+System.currentTimeMillis());
        timer.newTimeout((timeout) -> {
            System.out.println("task execute,current timestamp="+System.currentTimeMillis());
            countDownLatch.countDown();
        }, 2000, TimeUnit.MILLISECONDS);

        timer.newTimeout((timeout) -> {
            System.out.println("task execute,current timestamp="+System.currentTimeMillis());
            countDownLatch.countDown();
        }, 2000, TimeUnit.MILLISECONDS);
        timer.newTimeout((timeout) -> {
            System.out.println("task execute,current timestamp="+System.currentTimeMillis());
            countDownLatch.countDown();
        }, 2000, TimeUnit.MILLISECONDS);
        countDownLatch.await();
        timer.stop();

    }

关键部分就
创建时间轮
HashedWheelTimer timer = new HashedWheelTimer(1, TimeUnit.SECONDS, 16);
时间轮添加任务
timer.newTimeout()

相关概念解析

HashedWheelTimer 时间轮中包含了那些信息
在这里插入图片描述
一个时间轮主要包含了这么多信息
wheel 是一个数组,和时间轮的曹数一样大的一个数组,即存放每个时间刻度的任务
timeout 就是任务的包装类
Queue timeouts
Queue cancelledTimeouts
两个对立,方便存放任务和取消任务的两个独立。由于是一个work线程访问,因此是线程安全的

时间轮 运行时序图

在这里插入图片描述

源码



package io.netty.util;

import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.util.Collections;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;

import static io.netty.util.internal.StringUtil.simpleClassName;


public class HashedWheelTimer implements Timer {

    static final InternalLogger logger =
            InternalLoggerFactory.getInstance(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 ResourceLeakDetector<HashedWheelTimer> leakDetector = ResourceLeakDetectorFactory.instance()
            .newResourceLeakDetector(HashedWheelTimer.class, 1);

    private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");

    private final ResourceLeakTracker<HashedWheelTimer> leak;
    private final Worker worker = new Worker();
    private final Thread workerThread;

    public static final int WORKER_STATE_INIT = 0;
    public static final int WORKER_STATE_STARTED = 1;
    public static final int WORKER_STATE_SHUTDOWN = 2;
    @SuppressWarnings({ "unused", "FieldMayBeFinal" })
    private volatile int workerState; // 0 - init, 1 - started, 2 - shut down

    private final long tickDuration;
    private final HashedWheelBucket[] wheel;
    private final int mask;
    private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
    private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();
    private final Queue<HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue();
    private final AtomicLong pendingTimeouts = new AtomicLong(0);
    private final long maxPendingTimeouts;

    private volatile long startTime;

   

    /**
     * Creates a new timer.
     *
     * @param threadFactory        a {@link ThreadFactory} that creates a
     *                             background {@link Thread} which is dedicated to
     *                             {@link TimerTask} execution.
     *                             用来创建worker线程
     * @param tickDuration         the duration between tick
     *                             tick的时长,也就是指针多久转一格
     * @param unit                 the time unit of the {@code tickDuration}
     *                             tickDuration的时间单位
     * @param ticksPerWheel        the size of the wheel
     *                             一圈有几格
     * @param leakDetection        {@code true} if leak detection should be enabled always,
     *                             if false it will only be enabled if the worker thread is not
     *                             a daemon thread.
     *                             是否开启内存泄露检测
     * @param  maxPendingTimeouts  The maximum number of pending timeouts after which call to
     *                             {@code newTimeout} will result in
     *                             {@link java.util.concurrent.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 &lt;= 0
     */
    public HashedWheelTimer(
            ThreadFactory threadFactory,
            long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
            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.
        //创建时间轮基本的数据结构,一个数组。长度为不小于ticksPerWheel的最小2的n次方
        wheel = createWheel(ticksPerWheel);
        // 这是一个标示符,用来快速计算任务应该呆的格子。
        // 我们知道,给定一个deadline的定时任务,其应该呆的格子=deadline%wheel.length.但是%操作是个相对耗时的操作,所以使用一种变通的位运算代替:
        // 因为一圈的长度为2的n次方,mask = 2^n-1后低位将全部是1,然后deadline&mast == deadline%wheel.length
        // java中的HashMap也是使用这种处理方法
        mask = wheel.length - 1;

        // Convert tickDuration to nanos.
        this.tickDuration = unit.toNanos(tickDuration);

        // Prevent overflow.
        // 校验是否存在溢出。即指针转动的时间间隔不能太长而导致tickDuration*wheel.length>Long.MAX_VALUE
        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));
        }
        //创建worker线程
        workerThread = threadFactory.newThread(worker);
        //这里默认是启动内存泄露检测:当HashedWheelTimer实例超过当前cpu可用核数*4的时候,将发出警告
        leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;

        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();
            }
        }
    }

    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的值为不小于ticksPerWheel的最小2的n次方
        ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
        // 初始化wheel数组
        HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
        for (int i = 0; i < wheel.length; i ++) {
            wheel[i] = new HashedWheelBucket();
        }
        return wheel;
    }
    /**
     * 这里其实不建议使用这种方式,因为当ticksPerWheel的值很大的时候,
     * 这个方法会循环很多次,方法执行时间不稳定,效率也不够。推荐使用java8 HashMap的做法:
     */
    private static int normalizeTicksPerWheel(int ticksPerWheel) {
        int normalizedTicksPerWheel = 1;
        while (normalizedTicksPerWheel < ticksPerWheel) {
            normalizedTicksPerWheel <<= 1;
        }
        return normalizedTicksPerWheel;
    }

    /**
     * Starts the background thread explicitly.  The background thread will
     * start automatically on demand even if you did not call this method.
     *显示的启动后台线程。即使你没有调用这个方法,后台线程也将启动
     *
     * 启动时间轮。这个方法其实不需要显示的主动调用,因为在添加定时任务(newTimeout()方法)的时候会自动调用此方法。
     *这个是合理的设计,因为如果时间轮里根本没有定时任务,启动时间轮也是空耗资源
     *
     * public void start() {
     * 判断当前时间轮的状态,如果是初始化,则启动worker线程,启动整个时间轮;
     * 如果已经启动则略过;
     * 如果是已经停止,则报错
     *
     * 这里是一个Lock Free(无锁)的设计。因为可能有多个线程调用启动方法,这里使用AtomicIntegerFieldUpdater原子的更新时间轮的状态
     * @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.
        //等待worker线程初始化时间轮的启动时间 startTimeInitialized.countDown();
        while (startTime == 0) {
            try {
                startTimeInitialized.await();
            } catch (InterruptedException ignore) {
                // Ignore - it will be ready very soon.
            }
        }
    }

    /**
     *停止时间轮
     */
    @Override
    public Set<Timeout> stop() {
        // worker线程不能停止时间轮,也就是加入的定时任务,不能调用这个方法。
        // 不然会有恶意的定时任务调用这个方法而造成大量定时任务失效
        if (Thread.currentThread() == workerThread) {
            throw new IllegalStateException(
                    HashedWheelTimer.class.getSimpleName() +
                            ".stop() cannot be called from " +
                            TimerTask.class.getSimpleName());
        }
        // 尝试CAS替换当前状态为“停止:2”。如果失败,则当前时间轮的状态只能是“初始化:0”或者“停止:2”。直接将当前状态设置为“停止:2“
        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();
                if (leak != null) {
                    boolean closed = leak.close(this);
                    assert closed;
                }
            }

            return Collections.emptySet();
        }
        //中断worker线程
        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();
            if (leak != null) {
                boolean closed = leak.close(this);
                assert closed;
            }
        }
        return worker.unprocessedTimeouts();
    }

    /**
     * 添加定时任务
     */
    @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.
        //计算任务的deadline
        // 这里定时任务不是直接加到对应的格子中,而是先加入到一个队列里,
        // 然后等到下一个tick的时候,会从队列里取出最多100000个任务加入到指定的格子中
        long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        timeouts.add(timeout);
        System.out.println("wheel ============="+wheel.length+ wheel[0].head);
        return timeout;
    }

    /**
     * Returns the number of pending timeouts of this {@link Timer}.
     */
    public long pendingTimeouts() {
        return pendingTimeouts.get();
    }

    private static void reportTooManyInstances() {
        String resourceType = 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.");
    }

worker

 //Worker是时间轮的核心线程类。tick的转动,过期任务的处理都是在这个线程中处理的。
 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.只有所有任务的的deadline都是想对于这个时间点
         startTime = System.nanoTime();
         // 由于System.nanoTime()可能返回0,甚至负数。并且0是一个标示符,用来判断startTime是否被初始化,所以当startTime=0的时候,重新赋值为1
         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;
         }
         logger.info("work ==== run===startTime: {} ", startTime);
         // Notify the other threads waiting for the initialization at start().
         // 唤醒阻塞在start()的线程
         startTimeInitialized.countDown();
         //只要时间轮的状态为WORKER_STATE_STARTED,就循环的“转动”tick,循环判断响应格子中的到期任务
         do {
             /**
              * 首先 worker 线程会通过 waitForNextTick()方法根据时间轮的时间刻度等待一轮循环的开始,
              * 在默认情况下时间轮的时间刻度是 100ms,
              * 那么此处 worker 线程也将在这个方法中 sleep 相应的时间等待下一轮循环的开始。
              * 此处也决定了时间轮的定时任务时间精度。
              *
              * waitForNextTick方法主要是计算下次tick的时间, 然后sleep到下次tick
              * 返回值就是System.nanoTime() - startTime, 也就是Timer启动后到这次tick, 所过去的时间
              */
             final long deadline = waitForNextTick();
             logger.info("work ==== deadline: {} ", deadline);
             if (deadline > 0) { // 可能溢出或者被中断的时候会返回负数, 所以小于等于0不管
                 //获取tick对应的格子索引
                 int idx = (int) (tick & mask);
                 // 移除被取消的任务
                 processCancelledTasks();
                 HashedWheelBucket bucket =
                         wheel[idx];
                 /**
                  * 当 worker 线程经过相应时间间隔的 sleep 之后,也代表新的一轮调度开始。
                  * 此时,会通过 transferTimeoutsToBuckets()方法将之前刚刚加入到
                  * timeouts 队列中的定时任务放入到时间轮具体槽位上的链表中。
                  *
                  * 首先,在每一轮的调度中,最多只会从 timeouts 队列中定位到时间轮 100000 个定时任务,
                  * 这也是为了防止在这里耗时过久导致后面触发定时任务的延迟。
                  * 在这里会不断从 timeouts 队列中获取刚加入的定时任务。
                  */
                 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.
         // 这里应该是时间轮停止了,清除所有格子中的任务,并加入到未处理任务列表,以供stop()方法返回
         for (HashedWheelBucket bucket: wheel) {
             bucket.clearTimeouts(unprocessedTimeouts);
         }
         // 将还没有加入到格子中的待处理定时任务队列中的任务取出,
         // 如果是未取消的任务,则加入到未处理任务队列中,以供stop()方法返回
         for (;;) {
             HashedWheelTimeout timeout = timeouts.poll();
             logger.info("work ==== timeout{} ", timeout);
             if (timeout == null) {
                 break;
             }
             if (!timeout.isCancelled()) {
                 unprocessedTimeouts.add(timeout);
             }
         }
         // 处理取消的任务
         processCancelledTasks();
     }
     // 将newTimeout()方法中加入到待处理定时任务队列中的任务加入到指定的格子中
     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.
         // 每次tick只处理10w个任务,以免阻塞worker线程
         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;
             }
             // 计算任务需要经过多少个tick
             long calculated = timeout.deadline / tickDuration;
             // 计算任务的轮数
             timeout.remainingRounds = (calculated - tick) / wheel.length;
             //如果任务在timeouts队列里面放久了, 以至于已经过了执行时间,
             // 这个时候就使用当前tick, 也就是放到当前bucket, 此方法调用完后就会被执行.
             final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
             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)
      * sleep, 直到下次tick到来, 然后返回该次tick和启动时间之间的时长
      */
     private long waitForNextTick() {
         //下次tick的时间点, 用于计算需要sleep的时间
         long deadline = tickDuration * (tick + 1);

         for (;;) {
             /**
              *  计算需要sleep的时间, 之所以加999999后再除10000000, 是为了保证足够的sleep时间
              * 例如:当deadline - currentTime=2000002的时候,如果不加999999,则只睡了2ms,
              * 而2ms其实是未到达deadline这个时间点的,所有为了使上述情况能sleep足够的时间,加上999999后,会多睡1ms
              */
             final long currentTime = System.nanoTime() - startTime;
             long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
             //这里的意思应该是从时间轮启动到现在经过太长的时间(跨度大于292年...),以至于让long装不下,都溢出了
             if (sleepTimeMs <= 0) {
                 if (currentTime == Long.MIN_VALUE) {
                     return -Long.MAX_VALUE;
                 } else {
                     return currentTime;
                 }
             }

             // Check if we run on windows, as if thats the case we will need
             // to round the sleepTime as workaround for a bug that only affect
             // the JVM if it runs on windows.
             //
             // See https://github.com/netty/netty/issues/356
             // 这里是因为windows平台的定时调度最小单位为10ms,如果不是10ms的倍数,可能会引起sleep时间不准确
             if (PlatformDependent.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;
                 }
             }
         }
     }

     public Set<Timeout> unprocessedTimeouts() {
         return Collections.unmodifiableSet(unprocessedTimeouts);
     }
 }

HashedWheelTimeout

 /**
  * HashedWheelTimeout是一个定时任务的内部包装类,双向链表结构。
  * 会保存定时任务到期执行的任务、deadline、round等信息。
  */
 private static final class HashedWheelTimeout implements Timeout {
     //定义定时任务的3个状态:初始化、取消、过期
     private static final int ST_INIT = 0;
     private static final int ST_CANCELLED = 1;
     private static final int ST_EXPIRED = 2;
     //用来CAS方式更新定时任务状态
     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;

     // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
     // HashedWheelTimeout will be added to the correct HashedWheelBucket.
     // 离任务执行的轮数,当将次任务加入到格子中是计算该值,每过一轮,该值减一。
     long remainingRounds;

     // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
     // As only the workerThread will act on it there is no need for synchronization / volatile.
     //双向链表结构,由于只有worker线程会访问,这里不需要synchronization / volatile
     HashedWheelTimeout next;
     HashedWheelTimeout prev;

     // The bucket to which the timeout was added
     //定时任务所在的格子
     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.
         //这里只是修改状态为ST_CANCELLED,会在下次tick时,在格子中移除
         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.
         // 加入到时间轮的待取消队列,并在每次tick的时候,从相应格子中移除。
         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() {
         final long currentTime = System.nanoTime();
         long remaining = deadline - currentTime + timer.startTime;

         StringBuilder buf = new StringBuilder(192)
            .append(simpleClassName(this))
            .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();
     }
 }

HashedWheelBucket

    /**
     * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
     * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
     * extra object creation is needed.
     * HashedWheelBucket用来存放HashedWheelTimeout,结构类似于LinkedList。
     * 提供了expireTimeouts(long deadline)方法来过期并执行格子中的定时任务
     */
    private static final class HashedWheelBucket {
        // Used for the linked-list datastructure
        // 指向格子中任务的首尾
        private HashedWheelTimeout head;
        private HashedWheelTimeout tail;

        /**
         * Add {@link HashedWheelTimeout} to this bucket.
         * // 基础的链表添加操作
         */
        public void addTimeout(HashedWheelTimeout timeout) {
            assert timeout.bucket == null;
            timeout.bucket = this;
            if (head == null) {
                head = tail = timeout;
            } else {
                tail.next = timeout;
                timeout.prev = tail;
                tail = timeout;
            }
        }

        /**
         * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
         * 过期并执行格子中的到期任务,tick到该格子的时候,worker线程会调用这个方法,
         * 根据deadline和remainingRounds判断任务是否过期
         */
        public 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.
                        //如果round数已经为0,deadline却>当前格子的deadline,说放错格子了,这种情况应该不会出现
                        throw new IllegalStateException(String.format(
                                "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                } else if (timeout.isCancelled()) {
                    next = remove(timeout);
                } else {
                    //没有到期,轮数-1
                    timeout.remainingRounds --;
                }
                timeout = next;
            }
        }

        //基础的链表移除node操作
        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.
         */
        public void clearTimeouts(Set<Timeout> set) {
            for (;;) {
                HashedWheelTimeout timeout = pollTimeout();
                if (timeout == null) {
                    return;
                }
                if (timeout.isExpired() || timeout.isCancelled()) {
                    continue;
                }
                set.add(timeout);
            }
        }
        //链表的poll操作
        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;
        }
    }
}


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/448386.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

hot100:数组——49、53、55

49. 字母异位词分组 用hashmap存储&#xff0c;其中每个key&#xff0c;用这组异位词的排序后的字符串&#xff1b;value是这组异位词。比如“tea”和“ate”是一组异位词&#xff0c;他们的排序结果都是“aet”。 public List<List<String>> groupAnagrams(Stri…

交流电中的无功功率和有功功率,减少无功功率

有功&#xff0c;无功功率 从字面上理解就是做功和不做功的功率。不做功的是因为负载电路中有电感和电容的存在。 电容和电感的电压电流关系 设加在两端的电压都是 U U m a x s i n w t UU_{max}sinwt UUmax​sinwt 电容和电感两端电压电流的关系 电容两端的电压电流关系…

Javaee Spring基于XML的AOP开发

快速入门 1. 导入 AOP 相关坐标 2. 创建目标接口和目标类&#xff08;内部有切点&#xff09; 3. 创建切面类&#xff08;内部有增强方法&#xff09; 4. 将目标类和切面类的对象创建权交给 spring 5. 在 applicationContext.xml 中配置织入关系 6. 测试代码 项目…

【数据库】MySQL数据约束和表关系详解

目录 1.数据库约束 1.1约束类型 1.2NULL约束 1.3UNIQUE&#xff1a;唯一约束 1.4DEFAULT&#xff1a;默认值约束 1.5PRIMARY&#xff1a;主键约束 1.6FOREIGH KEY:外键约束 2.表的关系 2.1一对一 2.2一对多 2.3多对多 1.数据库约束 数据库中的数据保存在数据表中&am…

数据库基础篇 《9. 子查询》

目录 1. 需求分析与问题解决 1.1 实际问题 1.2 子查询的基本使用 ​编辑1.3 子查询的分类 分类方式1&#xff1a;我们按内查询的结果返回一条还是多条记录&#xff0c;将子查询分为 单行子查询 、 多行子查询 。 分类方式2&#xff1a; 我们按内查询是否被执行多次&#x…

4WE10D50型力士乐液压阀规格

安装位置 可选择&#xff0c;方向阀最好水平安装&#xff01;对于阀品种&#xff0c;如 &#xff0d; 不带阀芯对中弹簧 &#xff0d; 或带下垂电磁铁 其他安装位置能够导致功能失常或违反有关的技朮规定。 带泄油口的压力开关的安装位置必须选择&#xff0c;使它的泄油口…

Redis源码分析之网络模型

Redis网络模型 阅读源码的初衷Redis源码阅读 阅读源码的初衷 很多网上解释这个Redis为啥这么块&#xff1f;都会说Redis这么快的原因会有一个Redis才用了单线程&使用了多路io复用来检查io事件&#xff0c;单线程可以避免多线程对资源的竞争。如果我们使用了多线程那么就需…

Revit中栏杆扶手、坡道的绘制及插件太多问题

一、在Revit中栏杆与扶手的绘制方法有两种&#xff1a; ①绘制路径 ②放置在主体上 二、执行方式 功能区&#xff1a;“建筑”选项卡“楼梯坡道”面板“栏杆扶手”下拉菜单“绘制路径”。 三、绘制技巧 首先我们一起来看看如何设置栏杆扶手属性。 第一步&#xff1a;设置属…

【vue3学习系列】对比vue2生命周期做了哪些改变,vue3初学者快来看看

文章目录 前言官方生命周期图分析去除beforeCreate与createdsetup代替created其他钩子只是改了名称 剔除vue2后的生命周期图其他钩子函数keepalive错误捕获其他的一些钩子去官方文档看看即可 前言 看了下官方的生命周期的说明&#xff0c;感觉讲的不算太清晰&#xff0c;所以个…

C++ 内联函数(inline)

内联函数&#xff1a;就是在函数前加inline 让函数在调用的地方直接展开 可是内联函数有什么作用呢&#xff1f;&#xff1f; 我们都知道&#xff0c;如果调用一个函数的话&#xff0c;会建立栈帧&#xff0c;在建立栈帧的时候会进行压栈等一系列操作。 而内联函数会在调用的…

Nginx和tomcat反向代理

七层反向代理 实验准备&#xff1a;准备三台虚拟机 192.168.146.20 tomcat&#xff08;两个&#xff09; 192.168.146.30 tomcat 192.168.146.50 七层反向代理&#xff08;nginx&#xff09; 部署虚拟机192.168.146.20&#xff08;两个tomcat已部署完毕&#xff09; …

SwiftUI 4.0(iOS 16)极简实现一个美美哒的多选 Toggle 按钮组

概览 在 SwiftUI 4.0 之前&#xff0c;想要实现如下效果的多选/全选 Toggle 按钮组是要写不少行代码滴&#xff1a; 不过&#xff0c;在 iOS 16 之后我们仅用1行代码即可搞定以上所有&#xff01;在某些场合下这非常有用哦。 在本篇博文中&#xff0c;我们就来看看如何实现它…

2023年的深度学习入门指南(6) - 剪枝和量化

2023年的深度学习入门指南(6) - 剪枝和量化 从这一节开始&#xff0c;我们要准备一些技术专项了。因为目前大模型技术还在快速更新迭代中&#xff0c;各种库和实现每天都在不停出现。因为变化快&#xff0c;所以难免会遇到一些问题。对于细节有一定的把握能力起码可以做到出问…

5.1 数值微分

学习目标&#xff1a; 作为数值分析的基础内容&#xff0c;我建议你可以采取以下步骤来学习数值微分&#xff1a; 掌握微积分基础&#xff1a;数值微分是微积分中的一个分支&#xff0c;需要先掌握微积分基础知识&#xff0c;包括导数、极限、微分等。 学习数值微分的概念和方…

Scillus | 来吧!它可以大大简化你的Seurat分析流程哦!~(二)(高级可视化)

1写在前面 不知道大家那里天气热了没有&#xff0c;苦逼的我虽然“享受”着医院的恒温&#xff0c;但也并没有什么卵用&#xff0c;毕竟我只是个不可以生锈的“小螺丝”。&#x1f972; 上期介绍了Scillus包的基本功能&#xff0c;如何进行数据的预处理及其可视化。&#x1f92…

分享github上比较热门的ChatGPT项目,值得收藏

&#x1f517; 运行环境&#xff1a;chatGPT &#x1f6a9; 撰写作者&#xff1a;左手の明天 &#x1f947; 精选专栏&#xff1a;《python》 &#x1f525; 推荐专栏&#xff1a;《算法研究》 #### 防伪水印——左手の明天 #### &#x1f497; 大家好&#x1f917;&#x1f9…

Elasticsearch painless脚本教程(包含Java API和SpringDataElasticsearch调用脚本)

Elasticsearch painless脚本 1.什么是painless2.painless的特性3.使用kibana进行准备操作3.1 使用kibana创建索引和映射3.2 使用kibana添加测试数据 4.使用painless执行查询操作关于脚本查询须知4.1 字段查询脚本4.1 排序查询脚本 5.如何使用painless执行更新操作关于脚本查询须…

Spring Security 04 自定义认证

登录⽤户数据获取 SecurityContextHolder Spring Security 会将登录⽤户数据保存在 Session 中。但是&#xff0c;为了使⽤⽅便, Spring Security 在此基础上还做了⼀些改进&#xff0c;其中最主要的⼀个变化就是线程绑定。当⽤户登录成功后,Spring Security 会将登录成功的⽤户…

Node内置模块 【crypto加密模块】

文章目录 &#x1f31f;前言&#x1f31f;crypto加密模块&#x1f31f;Crypto模块介绍&#x1f31f;Hash算法&#x1f31f;Hash算法介绍&#x1f31f;Hash算法之MD5&#x1f31f;算法简介&#x1f31f;MD5加密使用 &#x1f31f;Hash算法之SHA1&#x1f31f;算法简介&#x1f3…

二叉树经典题题解

目录 &#x1f345;1.单值二叉树&#x1f345; &#x1f349; 2.相同的树&#x1f349; &#x1f34a;3.对称二叉树&#x1f34a; &#x1f34e;4.另一颗树的子树&#x1f34e; &#x1f34f;5.翻转二叉树&#x1f34f; &#x1f351;6.平衡二叉树&#x1f351; &#x1f3…