4 新增数据
先看看继承自 BlockingQueue 的方法
put
- 将指定的元素插入此延迟队列。 由于队列*,因此此方法将永远不会阻塞.
- 可以看到 put 调用的是 offer
DelayQueue#offer
- 将指定的元素插入此延迟队列
执行流程
1.加锁
2.元素添加到优先级队列中
3.检验元素是否为队首,是则设置 leader 为null, 并唤醒一个消费线程
4.解锁
其内部调用的是 PriorityQueue 的 offer 方法
PriorityQueue#offer
将指定的元素插入此优先级队列.
public boolean offer(E e) { // 若元素为 null,抛NPE if (e == null) throw new NullPointerException(); // 修改计数器加一 modCount++; int i = size; // 如果队列大小 > 容量 if (i >= queue.length) // => 扩容 grow(i + 1); size = i + 1; // 若队列空,则当前元素正好处于队首 if (i == 0) queue[0] = e; else // 若队列非空,根据优先级排序 siftUp(i, e); return true; }
执行流程
- 元素判空
- 队列扩容判断
- 根据元素的 compareTo 方法进行排序,希望最终排序的结果是从小到大的,因为想让队首的都是过期的数据,需要在 compareTo 方法实现.
5 取数据
take
检索并删除此队列的头,如有必要,请等待直到延迟过期的元素在此队列上可用
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; // 获取可中断锁 lock.lockInterruptibly(); try { for (;;) { // 从优先级队列中获取队首 E first = q.peek(); if (first == null) // 队首为 null,说明无元素,当前线程加入等待队列,并阻塞 available.await(); else { // 获取延迟时间 long delay = first.getDelay(NANOSECONDS); if (delay <= 0) // 已到期,获取并删除头部元素 return q.poll(); first = null; // 在等待时不要保留引用 if (leader != null) available.await(); else { Thread thisThread = Thread.currentThread(); leader = thisThread; try { // 线程节点进入等待队列 available.awaitNanos(delay); } finally { if (leader == thisThread) leader = null; } } } } } finally { // 若leader == null且还存在元素,则唤醒一个消费线程 if (leader == null && q.peek() != null) available.signal(); // 解锁 lock.unlock(); } }
执行流程
- 加锁
- 取出优先级队列的队首
- 若队列为空,阻塞
- 若队首非空,获得这个元素的delay时间值,如果first的延迟delay时间值为0的话,说明该元素已经到了可以使用的时间,调用poll方法弹出该元素,跳出方法
- 若first的延迟delay时间值非0,释放元素first的引用,避免内存泄露
- 循环以上操作,直至return
take 方法是会无限阻塞,直到队头的过期时间到了才会返回.
如果不想无限阻塞,可以尝试 poll 方法,设置超时时间,在超时时间内,队头元素还没有过期的> 话,就会返回 null.
6 解密 leader 元素
leader 是一个Thread元素,表示当前获取到锁的消费者线程.
- 以take代码段为例
- 若 leader 非 null,说明已有消费者线程获取锁,直接阻塞当前线程.
若 leader 为 null,把当前线程赋给 leader,并等待剩余的到期时间,最后释放 leader.
这里假设有多个消费者线程执行 take 取数据,若没有leader != null 判断,这些线程都会无限循环,直到返回第一个元素,这显然很浪费系统资源. 所以 leader 在这里相当于一个线程标识,避免消费者线程的无脑竞争.
注意这里因为first是队首的引用,阻塞时会有很多线程同时持有队首引用,可能导致内存溢出,所以需要手动释放.
7 总结
DelayQueue 使用排序和超时机制即实现了延迟队列.充分利用已有的 PriorityQueue 排序功能,超时阻塞又恰当好处的利用了锁的等待,在已有机制的基础上进行封装.在实际开发中,可以多多实践这一思想,使代码架构具备高复用性.