java -jar xxx.jar

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

// 元素必须实现 Delayed 接口
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> { // 同时也是一个阻塞队列

    private final transient ReentrantLock lock = new ReentrantLock();
    private final PriorityQueue<E> q = new PriorityQueue<E>(); // 优先队列,内部是数组方式实现
    ...
}

public boolean offer(E e) { // 插入元素
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        q.offer(e);
        if (q.peek() == e) {
            leader = null;
            available.signal();
        }
        return true;
    } finally {
        lock.unlock();
    }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            E first = q.peek();
            if (first == null)
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return q.poll();
                first = null; // don't retain ref while waiting
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {
    this.worker = new HashedWheelTimer.Worker();
    this.startTimeInitialized = new CountDownLatch(1);
    this.timeouts = PlatformDependent.newMpscQueue();
    this.cancelledTimeouts = PlatformDependent.newMpscQueue();
    this.pendingTimeouts = new AtomicLong(0L);
    ObjectUtil.checkNotNull(threadFactory, "threadFactory");
    ObjectUtil.checkNotNull(unit, "unit");
    ObjectUtil.checkPositive(tickDuration, "tickDuration");
    ObjectUtil.checkPositive(ticksPerWheel, "ticksPerWheel");
    this.wheel = createWheel(ticksPerWheel); // ticksPerWheel 默认512，即槽的数量
    this.mask = this.wheel.length - 1; // mask是为了位运算用的为了提高性能，限制wheel.length 为2的n次方，即 tick & (wheel.length - 1) = tick % wheel.length
    long duration = unit.toNanos(tickDuration); // 每隔的时间
    if (duration >= 9223372036854775807L / (long)this.wheel.length) {
        throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, 9223372036854775807L / (long)this.wheel.length));
    } else {
        if (duration < MILLISECOND_NANOS) {
            logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", tickDuration, MILLISECOND_NANOS);
            this.tickDuration = MILLISECOND_NANOS;
        } else {
            this.tickDuration = duration;
        }

        this.workerThread = threadFactory.newThread(this.worker); // 工作线程
        this.leak = !leakDetection && this.workerThread.isDaemon() ? null : leakDetector.track(this); // 内存泄漏探测
        this.maxPendingTimeouts = maxPendingTimeouts;
        if (INSTANCE_COUNTER.incrementAndGet() > 64 && WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
            reportTooManyInstances();
        }

    }
}

public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
    ObjectUtil.checkNotNull(task, "task");
    ObjectUtil.checkNotNull(unit, "unit");
    long pendingTimeoutsCount = this.pendingTimeouts.incrementAndGet();
    if (this.maxPendingTimeouts > 0L && pendingTimeoutsCount > this.maxPendingTimeouts) {
        this.pendingTimeouts.decrementAndGet();
        throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending timeouts (" + this.maxPendingTimeouts + ")");
    } else {
        // 开启时间轮轮询
        this.start();
        long deadline = System.nanoTime() + unit.toNanos(delay) - this.startTime;
        if (delay > 0L && deadline < 0L) {
            deadline = 9223372036854775807L;
        }
		// 将定时任务封装成HashedWheelTimeout
        HashedWheelTimer.HashedWheelTimeout timeout = new HashedWheelTimer.HashedWheelTimeout(this, task, deadline);
        // 将定时任务进入到MPSC队列中
        this.timeouts.add(timeout);
        return timeout;
    }
}

do {
    long deadline = this.waitForNextTick(); // 等待执行任务的时间到来
    if (deadline > 0L) {
        idx = (int)(this.tick & (long)HashedWheelTimer.this.mask);
        this.processCancelledTasks(); // 处理那些被取消了的任务，即取消队列poll，移除任务
        bucket = HashedWheelTimer.this.wheel[idx]; // 选择对应的槽
        this.transferTimeoutsToBuckets(); // 将加入 MPSC 队列的任务加入到时间轮的槽中
        bucket.expireTimeouts(deadline); // 处理时间到了的任务
        ++this.tick; // 记一次移动
    }
} while(HashedWheelTimer.WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == 1);

private long waitForNextTick() {
    long deadline = HashedWheelTimer.this.tickDuration * (this.tick + 1L);
    while(true) {
        long currentTime = System.nanoTime() - HashedWheelTimer.this.startTime;
        // +999999L 保证足够的sleep时间，比如 deadline - currentTime 为 5纳秒，
        // 那么5纳秒转毫秒就为0了，但是实际时间还没到
        long sleepTimeMs = (deadline - currentTime + 999999L) / 1000000L;
        if (sleepTimeMs <= 0L) { // 时间过了，马上执行任务
            if (currentTime == -9223372036854775808L) { // 溢出问题？
                return -9223372036854775807L;
            }
            return currentTime;
        }
        // 在windows下有bug，sleep时间是10的整数倍，
        // http://github.com/netty/netty/issues/356
        // 这个 bug 是由 windows 在 java 调用 Thread.sleep() 时产生中断引起的，如果时间低于 10 毫秒，则通过不休眠来解决。
        if (PlatformDependent.isWindows()) {
            sleepTimeMs = sleepTimeMs / 10L * 10L;
            if (sleepTimeMs == 0L) {
                sleepTimeMs = 1L;
            }
        }
        try {
            Thread.sleep(sleepTimeMs); // 等待时间的到来
        } catch (InterruptedException var8) {
            if (HashedWheelTimer.WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == 2) {
                return -9223372036854775808L;
            }
        }
    }
}

private void transferTimeoutsToBuckets() {
    for(int i = 0; i < 100000; ++i) {
        HashedWheelTimer.HashedWheelTimeout timeout = (HashedWheelTimer.HashedWheelTimeout)HashedWheelTimer.this.timeouts.poll(); // 获取任务
        if (timeout == null) {
            break;
        }
        if (timeout.state() != 1) {
            long calculated = timeout.deadline / HashedWheelTimer.this.tickDuration;
            timeout.remainingRounds = (calculated - this.tick) / (long)HashedWheelTimer.this.wheel.length; // 计算在第几轮
            long ticks = Math.max(calculated, this.tick);
            int stopIndex = (int)(ticks & (long)HashedWheelTimer.this.mask); // 计算在哪个槽
            HashedWheelTimer.HashedWheelBucket bucket = HashedWheelTimer.this.wheel[stopIndex];
            bucket.addTimeout(timeout); // 入槽
        }
    }
}

public void expireTimeouts(long deadline) {
    HashedWheelTimer.HashedWheelTimeout next; // 拿到槽的链表头部
    for(HashedWheelTimer.HashedWheelTimeout timeout = this.head; timeout != null; timeout = next) {
        next = timeout.next;
        if (timeout.remainingRounds <= 0L) {
            next = this.remove(timeout);
            if (timeout.deadline > deadline) {
                throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
            }
            timeout.expire(); // 执行，实际上就是调用任务的run方法
        } else if (timeout.isCancelled()) {
            next = this.remove(timeout);
        } else {
            --timeout.remainingRounds; // 否则轮数减一
        }
    }
}