java并发中DelayQueue延迟队列原理剖析

 public interface Delayed extends Comparable { /** * Returns the remaining delay associated with this object, in the * given time unit. * * @param unit the time unit * @return the remaining delay; zero or negative values indicate * that the delay has already elapsed */ long getDelay(TimeUnit unit); } 

 // 锁。 private final transient ReentrantLock lock = new ReentrantLock(); // 优先队列。 private final PriorityQueue q = new PriorityQueue(); /** * Leader-Follower的变种。 * Thread designated to wait for the element at the head of * the queue.  This variant of the Leader-Follower pattern * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to * minimize unnecessary timed waiting.  When a thread becomes * the leader, it waits only for the next delay to elapse, but * other threads await indefinitely.  The leader thread must * signal some other thread before returning from take() or * poll(...), unless some other thread becomes leader in the * interim.  Whenever the head of the queue is replaced with * an element with an earlier expiration time, the leader * field is invalidated by being reset to null, and some * waiting thread, but not necessarily the current leader, is * signalled.  So waiting threads must be prepared to acquire * and lose leadership while waiting. */ private Thread leader = null; /** * Condition signalled when a newer element becomes available * at the head of the queue or a new thread may need to * become leader. */ // 条件，代表如果有数据则通知Follower线程，唤醒线程处理队列内容。 private final Condition available = lock.newCondition(); 

 public boolean offer(E e) { // 获取到锁 final ReentrantLock lock = this.lock; lock.lock(); try { // 将元素存储到PriorityQueue优先队列中 q.offer(e); // 如果第一个元素是当前元素，说明之前队列中为空，则先将Leader设置为空，通知等待线程可以争抢Leader了。 if (q.peek() == e) { leader = null; available.signal(); } // 返回成功 return true; } finally { lock.unlock(); } } 

 public void put(E e) { offer(e); } 

 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); // 如果剩余过期时间小于0，则说明已经过期，反之还没有过期。 if (delay <= ) // 如果已经过期直接获取最早过期的元素，并返回。 return q.poll(); // 如果剩余过期日期大于0，则会进入到这里。 // 将刚才获取的最早过期的元素设置为空。 first = null; // don't retain ref while waiting // 如果有线程争抢的Leader线程，则进行无限期等待。 if (leader != null) // 无限期等待并让出锁。 available.await(); else { // 获取当前线程。 Thread thisThread = Thread.currentThread(); // 设置当前线程变为Leader线程。 leader = thisThread; try { // 等待剩余等待时间。 available.awaitNanos(delay); } finally { // 将Leader设置为null。 if (leader == thisThread) leader = null; } } } } } finally { // 如果队列不为空，并且没有Leader则通知等待线程可以成为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(); // 如果队列为空或者队列最早过期的元素没有过期，则返回null。 if (first == null || first.getDelay(NANOSECONDS) > 0) return null; else // 出队列操作。 return q.poll(); } finally { lock.unlock(); } }