java无界阻塞队列 DelayQueue

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("============结束");
    }
}