public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
           implements BlockingQueue<E> {
// 用于控制并发的 可重入 全局 锁
private final transient ReentrantLock lock = new ReentrantLock();
// 根据Delay时间排序的 无界的 优先级队列
private final PriorityQueue<E> q = new PriorityQueue<E>();
// 用于优化阻塞通知的线程元素leader，标记当前是否有线程在排队（仅用于取元素时）
private Thread leader = null;
// 条件，用于阻塞和通知的Condition对象，表示现在是否有可取的元素
private final Condition available = lock.newCondition();
 
       /**
        * 省洛方法代码.....  你们懂我的省洛吗?
        */
复制代码

• 注释的已经很清楚他们的意思了,也具备了并发编程之艺术的 锁,队列,状态(条件)
• 他的几个方法也是通过 锁-->维护队列-->出队,入队-->根据Condition进行条件的判断-->进行线程之间的通信和唤起
• 以支持优先级无界队列的PriorityQueue作为一个容器，容器里面的元素都应该实现Delayed接口，在每次往优先级队列中添加元素时以元素的过期时间作为排序条件，最先过期的元素放在优先级最高。
• DelayQueue是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。

2) 优先级队列 PriorityQueue

因为我们的DelayQueue里面维护了一个优先级的队列PriorityQueue 简单的看下:

    //默认容量11
     private static final int DEFAULT_INITIAL_CAPACITY = 11;
    //存储元素的地方 数组
    transient Object[] queue; // non-private to simplify nested class access
    //元素个数
    private int size = 0;
    //比较器
    private final Comparator<? super E> comparator;
复制代码

1. 默认容量是11；
2. queue，元素存储在数组中，这跟我们之前说的堆一般使用数组来存储是一致的；
3. comparator，比较器，在优先级队列中，也有两种方式比较元素，一种是元素的自然顺序，一种是通过比较器来比较；
4. modCount，修改次数，有这个属性表示PriorityQueue也是fast-fail的；
5. PriorityQueue不是有序的，只有堆顶存储着最小的元素；
6. PriorityQueue 是非线程安全的；

3) DelayQueue的方法简介

• 入队方法 : 若添加的元素是队首(堆顶)元素，就把leader置为空，并唤醒等待在条件available上的线程；

public boolean add(E e) {    return offer(e);}
public void put(E e) {    offer(e);}
public boolean offer(E e, long timeout, TimeUnit unit) {    return offer(e);}
public boolean offer(E e) {    
    final ReentrantLock lock = this.lock;    
    lock.lock();   //加锁 因为优先队列线程不安全
    try {
        q.offer(e);  //判断优先级 进行入队      
    if (q.peek() == e) {    //-----[1]
        //leader记录了被阻塞在等待队列头生效的线程 新增一个元素到队列头，
        //表示等待原来队列头生效的阻塞的线程已经失去了阻塞的意义
        //，此时需要获取新的队列头进行返回了
        leader = null;      
        //获取队列头的线程被唤起,主要有两种场景:
        //1. 之前队列为空，导致被阻塞的线程
        //2. 之前队列非空，但是队列头没有生效(到期)导致被阻塞的线程
        available.signal();     
    }        
        return true; //因为是无界队列 所以添加元素肯定成功  直到OOM
    } finally {    
        lock.unlock();   //释放锁
    }
}
复制代码

offer()方法,首先获取独占锁，然后添加元素到优先级队列，由于q是优先级队列，所以添加元素后，peek并不一定是当前添加的元素，如果[1]为true，说明当前元素e的优先级最小也就即将过期的，这时候激活avaliable变量条件队列里面的线程，通知他们队列里面有元素了。

• 出队方法 take()

请看我详细的注释,绝不是蜻蜓点水

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock; //获取锁 
    lock.lockInterruptibly();   //可中断锁 可以自行了解一下 嘻嘻嘻嘻...
    try {
        for (;;) {//会写死循环的都是高手
            E first = q.peek();//get队头元素
            if (first == null)
                // 队列头为空，则阻塞，直到新增一个入队为止(1)
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);//获取剩余时间
                if (delay <= 0)
                    // 若队列头元素已生效，则直接返回(2)
                    return q.poll();
                first = null; // don't retain ref while waiting 等待的时候不能引用,表示释放当前引用的(3)
                if (leader != null)
                    // leader 非空时，表示有其他的一个线程在出队阻塞中 (4.1)
                    // 此时挂住当前线程，等待另一个线程出队完成
                    available.await();
                else {
                    //标识当前线程处于等待队列头生效的阻塞中 (4.2.1)
                    Thread thisThread = Thread.currentThread(); 
                    leader = thisThread;
                    try {
                        // 等待队列头元素生效(4.2.2)
                        available.awaitNanos(delay);
                    } finally {
                        //最终释放当前的线程 设置leader为null (4.2.3)
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }     //(5)
    } finally {
        if (leader == null && q.peek() != null)
            // 当前线程出队完成，通知其他出队阻塞的线程继续执行(6)
            available.signal();
            lock.unlock();//解锁结束
    }
}
复制代码

那么，下面的结论肉眼可见:

1. 如果队列为空，则阻塞，直到有个线程(生产者投递数据)完成入队操作
2. 获取队列头，若队列头已生效，则直接返回
3. 未生效则释放当前引用
4. 当队列头部没有生效时候:
   1. 若有另一个线程已经处于等待队列头生效的阻塞过程中，则阻塞当前线程，直到另一个线程完成出队操作
   2. 若没有其他线程阻塞在出队过程中，即当前线程为第一个获取队列头的线程
      • 标识当前线程处于等待队列头生效的阻塞中（leader = thisThread）
      • 阻塞当前线程，等待队列头生效
      • 队列头生效之后，清空标识(leader=null)
5. 再次进入循环，获取队列头并返回
6. 最后,当前线程出队完成，通知其他出队阻塞的线程继续执行

4)Delayed

public interface Delayed extends Comparable<Delayed> { 
    long getDelay(TimeUnit unit);
}
复制代码

据情报显示:Delayed是一个继承自Delayed的接口，并且定义了一个Delayed方法，用于表示还有多少时间到期，到期了应返回小于等于0的数值。

很简答就是定义了一个,一个哈,一个表延迟的接口,就是个规范接口,目的就是骗我们去实现它的方法.哼~

五、后总结

• 这就是单机的不好处,也是一个痛点,所以肯定是不太适合订单量特别大的场景 大家也要酌情考虑和运用
• 相对于同等量级的数据库轮询操作来说,真是节省了不少数据库的压力和连接,还是值得一用的,我们可以只保存订单的id到延时实例中,这样缩减队列单个实例内存存储
• 那还有技巧就是更新的时候注意控制好幂等性,控制好幂等性,会让你轻松很多,顺畅很多,但是数据量大了,要蛀牙的哦