1.官方文档
An unbounded blocking queue of Delayed elements, in which an
element can only be taken when its delay has expired. The head of
the queue is that Delayed element whose delay expired furthest in
the past. If no delay has expired there is no head and poll will return
null. Expiration occurs when an element's
getDelay(TimeUnit.NANOSECONDS) method returns a value less
than or equal to zero. Even though unexpired elements cannot be
removed using take or poll, they are otherwise treated as normal
elements. For example, the size method returns the count of both
expired and unexpired elements. This queue does not permit null
elements.
存放Delayed元素的无界阻塞队列，只有元素过期才能将其取走。队头head是过期最长时间的元素，如果没有元素过期则没有head并且poll会返回null。当元素的getDelay()方法返回小于等于0的值，则过期。不允许null元素。
2.Delayed
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
元素必须实现Delayed接口，方法getDelay返回元素还剩多少时间才过时。
由于Delayed继承自 Comparable<Delayed>，所以也要实现compareTo方法。
public interface Delayed extends Comparable<Delayed> {
/**
* 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);
}
3.优先队列
将元素入队时，会调用Delayed元素的compareTo方法进行排序。因此，元素除了提供getDelay，还需要提供compareTo方法。
private final PriorityQueue<E> q = new PriorityQueue<E>();
4.入队和出队
4.1 阻塞版本put和take
由于是无界队列，入队不会阻塞。
/**
* Inserts the specified element into this delay queue. As the queue is
* unbounded this method will never block.
*
* @param e the element to add
* @throws NullPointerException {@inheritDoc}
*/
public void put(E e) {
offer(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();
}
}
为什么q.peek() == e才signal？
当队列头部元素被更早到期的元素替换是，leader被置为null，offer里面q.peek() == e时，会将leader=null，此时当然会signal，重新竞选leader。所以定时等待线程必须要处理失去leader时情况。
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();
}
}
leader的作用是什么？
声明 private Thread leader = null;源码注释：
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.
leader是等待队列头部元素的指定线程。Leader-Follower模式的这种变体用于最小化不必要的定时等待。
- 当一个线程称为leader时，其会定时等待下一个delay元素过期，但是其他线程会无限期等待。
- 当从take/poll返回之前，leader线程必须signal其他等待线程，除非在此期间有线程称为了新的leader。
- 每当队列头部元素被更早到期的元素替换时，leader被置为null，offer里面q.peek() == e时，会将leader=null，此时当然会signal，重新竞选leader。所以定时等待线程必须要处理失去leader时情况。
4.2 返回特殊值的offer和poll（还有超时版本）
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 boolean offer(E e, long timeout, TimeUnit unit) {
return offer(e);
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
if (first == null) {
if (nanos <= 0)
return null;
else
nanos = available.awaitNanos(nanos);
} else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();
if (nanos <= 0)
return null;
first = null; // don't retain ref while waiting
if (nanos < delay || leader != null)
nanos = available.awaitNanos(nanos);
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
long timeLeft = available.awaitNanos(delay);
nanos -= delay - timeLeft;
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
4.3 抛出异常的add和remove
public boolean add(E e) {
return offer(e);
}
public boolean remove(Object o) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return q.remove(o);
} finally {
lock.unlock();
}
}
5.Leader-Follower模式
参考Concurrency Pattern
参考Leader-Follower线程模型简介
- 1）线程有3种状态：领导leading，处理processing，追随following
- 2）假设共N个线程，其中只有1个leading线程（等待任务），x个processing线程（处理），余下有N-1-x个following线程（空闲）
- 3）有一把锁，谁抢到就是leading
- 4）事件/任务来到时，leading线程会对其进行处理，从而转化为processing状态，处理完成之后，又转变为following
- 5）丢失leading后，following会尝试抢锁，抢到则变为leading，否则保持following
- 6）following不干事，就是抢锁，力图成为leading
