DelayQueue延迟队列原理剖析
Posted BattleHeart
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了DelayQueue延迟队列原理剖析相关的知识，希望对你有一定的参考价值。
DelayQueue延迟队列原理剖析
介绍
DelayQueue队列是一个延迟队列，DelayQueue中存放的元素必须实现Delayed接口的元素，实现接口后相当于是每个元素都有个过期时间，当队列进行take获取元素时，先要判断元素有没有过期，只有过期的元素才能出队操作，没有过期的队列需要等待剩余过期时间才能进行出队操作。
源码分析
DelayQueue队列内部使用了PriorityQueue优先队列来进行存放数据，它采用的是二叉堆进行的优先队列，使用ReentrantLock锁来控制线程同步，由于内部元素是采用的PriorityQueue来进行存放数据，所以Delayed接口实现了Comparable接口，用于比较来控制优先级，如下代码所示：
1public interface Delayed extends Comparable<Delayed> {
2
3 /**
4 * Returns the remaining delay associated with this object, in the
5 * given time unit.
6 *
7 * @param unit the time unit
8 * @return the remaining delay; zero or negative values indicate
9 * that the delay has already elapsed
10 */
11 long getDelay(TimeUnit unit);
12}
DelayQueue的成员变量如下所示：
1// 锁。
2private final transient ReentrantLock lock = new ReentrantLock();
3// 优先队列。
4private final PriorityQueue<E> q = new PriorityQueue<E>();
5
6/**
7 * Leader-Follower的变种。
8 * Thread designated to wait for the element at the head of
9 * the queue. This variant of the Leader-Follower pattern
10 * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
11 * minimize unnecessary timed waiting. When a thread becomes
12 * the leader, it waits only for the next delay to elapse, but
13 * other threads await indefinitely. The leader thread must
14 * signal some other thread before returning from take() or
15 * poll(...), unless some other thread becomes leader in the
16 * interim. Whenever the head of the queue is replaced with
17 * an element with an earlier expiration time, the leader
18 * field is invalidated by being reset to null, and some
19 * waiting thread, but not necessarily the current leader, is
20 * signalled. So waiting threads must be prepared to acquire
21 * and lose leadership while waiting.
22 */
23private Thread leader = null;
24
25/**
26 * Condition signalled when a newer element becomes available
27 * at the head of the queue or a new thread may need to
28 * become leader.
29 */
30// 条件，代表如果有数据则通知Follower线程，唤醒线程处理队列内容。
31private final Condition available = lock.newCondition();
Leader-Follower模式的变种，用于最小化不必要的定时等待，当一个线程被选择为Leader时，它会等待延迟过去执行代码逻辑，而其他线程则需要无限期等待，在从take或poll返回之前，每当队列的头部被替换为具有更早到期时间的元素时，leader字段将通过重置为空而无效，Leader线程必须向其中一个Follower线程发出信号，被唤醒的 follwer 线程被设置为新的Leader 线程。
offer操作
1public boolean offer(E e) {
2 // 获取到锁
3 final ReentrantLock lock = this.lock;
4 lock.lock();
5 try {
6 // 将元素存储到PriorityQueue优先队列中
7 q.offer(e);
8 // 如果第一个元素是当前元素，说明之前队列中为空，则先将Leader设置为空，通知等待线程可以争抢Leader了。
9 if (q.peek() == e) {
10 leader = null;
11 available.signal();
12 }
13 // 返回成功
14 return true;
15 } finally {
16 lock.unlock();
17 }
18}
offer操作前先进行获取锁的操作，也就是同一时间内只能有一个线程可以入队操作。
- 获取到ReentrantLock锁对象。
- 将元素添加到PriorityQueue优先队列中
- 如果队列中最早过期的元素是自己，则说明队列原先是空的，所以将Leader进行重置，通知Follower线程可以成为Leader线程。
- 最后进行解锁操作。
put操作
put操作其实就是调用的offer操作来进行添加数据的，以下是源码信息：
1public void put(E e) {
2 offer(e);
3}
take操作
1public E take() throws InterruptedException {
2 final ReentrantLock lock = this.lock;
3 // 获取可中断的锁。
4 lock.lockInterruptibly();
5 try {
6 // 循环获取数据。
7 for (;;) {
8 // 获取最早过期的元素，但是不弹出对象。
9 E first = q.peek();
10 // 如果最早过期的元素为空，说明队列为空，则线程直接进入无限期等待，并且让出锁。
11 if (first == null)
12 // 当前线程无限期等待，直到被唤醒，并且让出锁对象。
13 available.await();
14 else {
15 // 获取最早过期的元素剩余过期时间。
16 long delay = first.getDelay(NANOSECONDS);
17 // 如果剩余过期时间小于0，则说明已经过期，反之还没有过期。
18 if (delay <= 0)
19 // 如果已经过期直接获取最早过期的元素，并返回。
20 return q.poll();
21 // 如果剩余过期日期大于0，则会进入到这里。
22 // 将刚才获取的最早过期的元素设置为空。
23 first = null; // don\'t retain ref while waiting
24 // 如果有线程争抢的Leader线程，则进行无限期等待。
25 if (leader != null)
26 // 无限期等待并让出锁。
27 available.await();
28 else {
29 // 获取当前线程。
30 Thread thisThread = Thread.currentThread();
31 // 设置当前线程变为Leader线程。
32 leader = thisThread;
33 try {
34 // 等待剩余等待时间。
35 available.awaitNanos(delay);
36 } finally {
37 // 将Leader设置为null。
38 if (leader == thisThread)
39 leader = null;
40 }
41 }
42 }
43 }
Java Review - 并发编程_DelayQueue原理&源码剖析