Java中的延迟队列DelayQueue源码解析

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 (;;) {
        	//从优先堆中获取堆顶元素，即优先级最高，即预定时间最近的元素
            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)
                	//若已经有leader了，则进入条件队列无限期等待
                    available.await();
                else {
                	//否则成为leader进入条件队列超时等待，到预期时间或者有更近时间元素插入就到同步队列竞争锁，再重复循环去取堆顶元素
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
    	//取完元素后若leader为null且队列中还有元素则从条件等待队列通知一个线程到同步队列
    	//为什么存在leader不为null的情况：leader线程从awaitNanos()中结束后没有竞争过新进take()的线程，因此继续在同步队列中被阻塞，因此无需再从条件等待队列中通知线程，直接让leader线程再去竞争锁，
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();//释放锁资源让同步队列中的线程竞争锁
    }
}

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return q.poll();
    } finally {
        lock.unlock();
    }
}

public int size() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return q.size();
    } finally {
        lock.unlock();
    }
}

public class DelayQueueTest {
    //队列元素类
    static class DelayTask implements Delayed {
        long exeTime;//预定执行时间
        public DelayTask(long exeTime) {
            this.exeTime = exeTime;
        }
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.exeTime - System.currentTimeMillis(), unit);
        }
        @Override
        public int compareTo(Delayed o) {
            long delta = getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
            return (int) delta;
        }
    }
    public static void main(String[] args) {
        DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
        for (int i = 1;i <= 5;i++) {
            delayQueue.offer(new DelayTask(System.currentTimeMillis() + new Random().nextInt(10)*1000));
        }
        for (int i = 1;i <= 3;i++) {
            new Thread(() -> {
                try {
                    while (true) {
                        DelayTask task = delayQueue.take();
                        System.out.printf("取出任务！取出时间：%s 任务预定执行时间：%s%n", hms(System.currentTimeMillis()), hms(task.exeTime));
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
    public static String hms(long milliseconds) {
        return new SimpleDateFormat("HH:mm:ss").format(milliseconds);
    }
}