一文带你深入了解Java中延时任务的实现

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

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 (;;) {
                // 获取队列第一个元素，如果队列为空
                // 阻塞住直到有新元素加入队列，offer等方法调用signal唤醒线程
                E first = q.peek();
                if (first == null)
                    available.await();
                else {
                    // 如果队列中有元素
                    long delay = first.getDelay(NANOSECONDS);
                    // 判断延时时间，如果到时间了，直接取出数据并return
                    if (delay <= 0)
                        return q.poll();
                    first = null;
                    // 如果leader为空则阻塞
                    if (leader != null)
                        available.await();
                    else {
                        // 获取当前线程
                        Thread thisThread = Thread.currentThread();
                        // 设置leader为当前线程
                        leader = thisThread;
                        try {
                            // 阻塞延时时间
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

@nonthreadsafe
    private[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) {

    private[this] val interval = tickMs * wheelSize
    private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) }

    private[this] var currentTime = startMs - (startMs % tickMs)
    
    @volatile private[this] var overflowWheel: TimingWheel = null

    private[this] def addOverflowWheel(): Unit = {
        synchronized {
        if (overflowWheel == null) {
            overflowWheel = new TimingWheel(
            tickMs = interval,
            wheelSize = wheelSize,
            startMs = currentTime,
            taskCounter = taskCounter,
            queue
            )
        }
        }
    }

    def add(timerTaskEntry: TimerTaskEntry): Boolean = {
        val expiration = timerTaskEntry.expirationMs

        if (timerTaskEntry.cancelled) {
        false
        } else if (expiration < currentTime + tickMs) {
        false
        } else if (expiration < currentTime + interval) {
        val virtualId = expiration / tickMs
        val bucket = buckets((virtualId % wheelSize.toLong).toInt)
        bucket.add(timerTaskEntry)

        if (bucket.setExpiration(virtualId * tickMs)) {
            queue.offer(bucket)
        }
        true
        } else {
        if (overflowWheel == null) addOverflowWheel()
        overflowWheel.add(timerTaskEntry)
        }
    }

    def advanceClock(timeMs: Long): Unit = {
        if (timeMs >= currentTime + tickMs) {
        currentTime = timeMs - (timeMs % tickMs)

        if (overflowWheel != null) overflowWheel.advanceClock(currentTime)
        }
    }
    }