package java.util.concurrent;
//基于PriorityQueue实现的，支持延时获取元素的无界阻塞队列，在创建元素时可
//指定多久才能从队列中获取当前元素，只有在延时期满时才能从队列中获取元素
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E>
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}
1、常用方法
构造方法
//可重入锁
private final transient ReentrantLock lock = new ReentrantLock();
//存储元素的优先级队列
private final PriorityQueue<E> q = new PriorityQueue<E>();
//用于减少线程的竞争，表示当前有线程正在获取队头元素
//leader 就是一个信号，告诉其它线程：你们不要再去获取元素了，它们延迟时间还没到期，
// 我都还没有取到数据呢，你们要取数据，等我取了再说
private Thread leader = null;
//条件控制，表示是否可以从队列中取数据
//Condition 条件在阻塞时会释放锁，在被唤醒时会再次获取锁，获取成功才会返回
private final Condition available = lock.newCondition();
public DelayQueue() {
}
/**
* addAll 在AbstractQueue中实现
*/
public DelayQueue(Collection<? extends E> c) {
this.addAll(c);
}
入队方法
public boolean add(E e) {
return offer(e);
}
/**
*/
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//通过PriorityQueue 入队
q.offer(e);
//peek 是获取的队头元素
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
public void put(E e) {
offer(e);
}
出队方法
/**
* 获取并移除队列的头元素，如果此队列为空，或者未到获取时间则返回 null
*/
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
//peek 是获取的队头元素
E first = q.peek();
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return q.poll();
} finally {
lock.unlock();
}
}
/**
* 获取并移除队列的头部元素，在指定的等待时间前 阻塞等待
*/
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;//达到超时指定时间，返回null
else
//未超时，在available条件上进行等待nanos时间
nanos = available.awaitNanos(nanos);
} else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();//延时到期，返回出队元素
if (nanos <= 0)
return null;//达到超时指定时间，返回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();
}
}
/**
*
*/
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();
}
}
其它
/**
* 返回队头元素，不出队
*/
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return q.peek();
} finally {
lock.unlock();
}
}
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return q.size();
} finally {
lock.unlock();
}
}
2、使用举例
public class XM implements Delayed {
private int age;
private long removeTime;
@Override
public String toString() {
return "XM{" +
"age=" + age +
'}';
}
public XM(int age) {
this.age = age;
removeTime = System.currentTimeMillis() + age * 1000;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(removeTime - System.currentTimeMillis(), unit);
}
@Override
public int compareTo(Delayed o) {
XM xmy = (XM) o;
//升序
return this.age - xmy.getAge();
}
public static void main(String[] args) throws Exception {
XM xm1 = new XM(5);
XM xm2 = new XM(10);
DelayQueue<XM> delayQueue = new DelayQueue<>();
delayQueue.add(xm1);
delayQueue.add(xm2);
System.out.println("============开始");
XM take = delayQueue.take();
System.out.println(take);
XM take2 = delayQueue.take();
System.out.println(take2);
System.out.println("============结束");
}
}
