并发队列之DelayQueue

package com.example.demo.study;import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;import lombok.Data;public class Study0210 {@Datastatic class MyDelayed implements Delayed {private long delayTime;//该任务需要再队列中的延迟的时候private long expire;//这个时间表示当前时间和延迟时间相加，这里就叫做到期时间private String taskName;//任务的名称public MyDelayed(long delayTime, String taskName) {this.delayTime = delayTime;this.taskName = taskName;this.expire = System.currentTimeMillis()+delayTime;}//指定优先级队列里面的比较规则，就跟上篇博客中说的优先级队列中说的比较器一样
        @Overridepublic int compareTo(Delayed o) {return (int)(this.getDelay(TimeUnit.MILLISECONDS)-o.getDelay(TimeUnit.MILLISECONDS));}//这个方法表示该任务在队列中还有多少剩余时间，也就是expire-当前时间
        @Overridepublic long getDelay(TimeUnit unit) {return unit.convert(this.expire-System.currentTimeMillis(), TimeUnit.MILLISECONDS);}}public static void main(String[] args) throws InterruptedException {//创建延迟队列DelayQueue<MyDelayed> queue = new DelayQueue<MyDelayed>();//创建任务丢到队列中Random random = new Random();for (int i = 1; i < 11; i++) {MyDelayed myDelayed = new MyDelayed(random.nextInt(500),"task"+i);queue.add(myDelayed);}//获取队列中的任务，这里只会跟超时时间最小的有关，和入队顺序无关MyDelayed myDelayed = queue.take();while(myDelayed!=null) {System.out.println(myDelayed.toString());myDelayed = queue.take();}}
}

//由此可是这个队列中存放的任务必须是Delayed类型的
public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {//独占锁private final transient ReentrantLock lock = new ReentrantLock();//优先级队列private final PriorityQueue<E> q = new PriorityQueue<E>();//leader线程，实际上每次进行入队和出队操作的只能是leader线程，其余的都叫做fallower线程，这里会用到一个leader-follower模式private Thread leader = null;//条件变量private final Condition available = lock.newCondition();//省略很多代码
}

public boolean offer(E e) {final ReentrantLock lock = this.lock;//获取锁
    lock.lock();try {//往优先级队列中添加一个元素
        q.offer(e);//注意，peek方法只是获取优先级队列中第一个元素，并不会删除//如果优先级队列中取的元素就是和当前添加的元素一样，说明当前元素就是达到过期要求的，于是设置leader线程为null//然后通知条件队列中的线程优先级队列中已经有元素了，可以过来取了if (q.peek() == e) {leader = null;available.signal();}return true;} finally {//释放锁
        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 {//如果优先级队列中有元素，那么肯定能走到这里来，然后取到该元素的超时时间，如果小于0，说明已经达到要求了，可以获取并删除队列中的元素long delay = first.getDelay(NANOSECONDS);if (delay <= 0)return q.poll();first = null; // don't retain ref while waiting//如果leader队列不为空，说明有其他线程正在执行take，于是就把当前线程放到条件队列中if (leader != null)available.await();//到这里，说明优先级队列中没有元素到超时时间，而且此时没有其他线程调用take方法，于是就把leader线程设置为当前线程，//然后当前leader线程就会等待一定的时间，等优先级队列中最快超时的元素；//在等待的时候，leader线程会释放锁，这时其他线程B可以调用offer方法添加元素，线程C也可以调用take方法，然后线程C就会在//上面这里阻塞无限长的时间，直到被唤醒else {Thread thisThread = Thread.currentThread();leader = thisThread;try {available.awaitNanos(delay);} finally {//当前线程阻塞一定时间之后，不管成功了没有，都会把leader线程重置为null，然后重新循环if (leader == thisThread)leader = null;}}}}//这里的意思就是当前线程移除元素成功之后，唤醒条件队列中的线程去继续从队列中获取元素} finally {if (leader == null && q.peek() != null)available.signal();//释放锁
        lock.unlock();}
}

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {//尝试获取队头元素，如果队头元素为空或者该延迟过期时间还没到，就返回nullE first = q.peek();if (first == null || first.getDelay(NANOSECONDS) > 0)return null;else//否则就获取并移除队头元素return q.poll();} finally {lock.unlock();}
}