ICode9

精准搜索请尝试: 精确搜索
首页 > 其他分享> 文章详细

DelayQueue延迟队列原理剖析

2021-06-07 23:05:19  阅读:178  来源: 互联网

标签:队列 lock 元素 剖析 过期 线程 DelayQueue Leader


DelayQueue延迟队列原理剖析

介绍

DelayQueue队列是一个延迟队列,DelayQueue中存放的元素必须实现Delayed接口的元素,实现接口后相当于是每个元素都有个过期时间,当队列进行take获取元素时,先要判断元素有没有过期,只有过期的元素才能出队操作,没有过期的队列需要等待剩余过期时间才能进行出队操作。

源码分析

DelayQueue队列内部使用了PriorityQueue优先队列来进行存放数据,它采用的是二叉堆进行的优先队列,使用ReentrantLock锁来控制线程同步,由于内部元素是采用的PriorityQueue来进行存放数据,所以Delayed接口实现了Comparable接口,用于比较来控制优先级,如下代码所示:

 1public interface Delayed extends Comparable<Delayed> {
2
3    /**
4     * Returns the remaining delay associated with this object, in the
5     * given time unit.
6     *
7     * @param unit the time unit
8     * @return the remaining delay; zero or negative values indicate
9     * that the delay has already elapsed
10     */
11    long getDelay(TimeUnit unit);
12}

DelayQueue的成员变量如下所示:

 1// 锁。
2private final transient ReentrantLock lock = new ReentrantLock();
3// 优先队列。
4private final PriorityQueue<E> q = new PriorityQueue<E>();
5
6/**
7 * Leader-Follower的变种。
8 * Thread designated to wait for the element at the head of
9 * the queue.  This variant of the Leader-Follower pattern
10 * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
11 * minimize unnecessary timed waiting.  When a thread becomes
12 * the leader, it waits only for the next delay to elapse, but
13 * other threads await indefinitely.  The leader thread must
14 * signal some other thread before returning from take() or
15 * poll(...), unless some other thread becomes leader in the
16 * interim.  Whenever the head of the queue is replaced with
17 * an element with an earlier expiration time, the leader
18 * field is invalidated by being reset to null, and some
19 * waiting thread, but not necessarily the current leader, is
20 * signalled.  So waiting threads must be prepared to acquire
21 * and lose leadership while waiting.
22 */
23private Thread leader = null;
24
25/**
26 * Condition signalled when a newer element becomes available
27 * at the head of the queue or a new thread may need to
28 * become leader.
29 */
30// 条件,代表如果有数据则通知Follower线程,唤醒线程处理队列内容。
31private final Condition available = lock.newCondition();

Leader-Follower模式的变种,用于最小化不必要的定时等待,当一个线程被选择为Leader时,它会等待延迟过去执行代码逻辑,而其他线程则需要无限期等待,在从take或poll返回之前,每当队列的头部被替换为具有更早到期时间的元素时,leader字段将通过重置为空而无效,Leader线程必须向其中一个Follower线程发出信号,被唤醒的 follwer 线程被设置为新的Leader 线程。

offer操作

 1public boolean offer(E e) {
2    // 获取到锁
3    final ReentrantLock lock = this.lock;
4    lock.lock();
5    try {
6        // 将元素存储到PriorityQueue优先队列中
7        q.offer(e);
8        // 如果第一个元素是当前元素,说明之前队列中为空,则先将Leader设置为空,通知等待线程可以争抢Leader了。
9        if (q.peek() == e) {
10            leader = null;
11            available.signal();
12        }
13        // 返回成功
14        return true;
15    } finally {
16        lock.unlock();
17    }
18}

offer操作前先进行获取锁的操作,也就是同一时间内只能有一个线程可以入队操作。

  1. 获取到ReentrantLock锁对象。
  2. 将元素添加到PriorityQueue优先队列中
  3. 如果队列中最早过期的元素是自己,则说明队列原先是空的,所以将Leader进行重置,通知Follower线程可以成为Leader线程。
  4. 最后进行解锁操作。

put操作

put操作其实就是调用的offer操作来进行添加数据的,以下是源码信息:

1public void put(E e) {
2    offer(e);
3}

take操作

 1public E take() throws InterruptedException {
2    final ReentrantLock lock = this.lock;
3    // 获取可中断的锁。
4    lock.lockInterruptibly();
5    try {
6        // 循环获取数据。
7        for (;;) {
8            // 获取最早过期的元素,但是不弹出对象。
9            E first = q.peek();
10            // 如果最早过期的元素为空,说明队列为空,则线程直接进入无限期等待,并且让出锁。
11            if (first == null)
12                // 当前线程无限期等待,直到被唤醒,并且让出锁对象。
13                available.await();
14            else {
15                // 获取最早过期的元素剩余过期时间。
16                long delay = first.getDelay(NANOSECONDS);
17                // 如果剩余过期时间小于0,则说明已经过期,反之还没有过期。
18                if (delay <= 0)
19                    // 如果已经过期直接获取最早过期的元素,并返回。
20                    return q.poll();
21                // 如果剩余过期日期大于0,则会进入到这里。
22                // 将刚才获取的最早过期的元素设置为空。
23                first = null; // don't retain ref while waiting
24                // 如果有线程争抢的Leader线程,则进行无限期等待。
25                if (leader != null)
26                    // 无限期等待并让出锁。
27                    available.await();
28                else {
29                    // 获取当前线程。
30                    Thread thisThread = Thread.currentThread();
31                    // 设置当前线程变为Leader线程。
32                    leader = thisThread;
33                    try {
34                        // 等待剩余等待时间。
35                        available.awaitNanos(delay);
36                    } finally {
37                        // 将Leader设置为null。
38                        if (leader == thisThread)
39                            leader = null;
40                    }
41                }
42            }
43        }
44    } finally {
45        // 如果队列不为空,并且没有Leader则通知等待线程可以成为Leader。
46        if (leader == null && q.peek() != null)
47            // 通知等待线程。
48            available.signal();
49        lock.unlock();
50    }
51}
  1. 当获取元素时,先获取到锁对象。
  2. 获取最早过期的元素,但是并不从队列中弹出元素。
  3. 最早过期元素是否为空,如果为空则直接让当前线程无限期等待状态,并且让出当前锁对象。
  4. 如果最早过期的元素不为空
  • 获取最早过期元素的剩余过期时间,如果已经过期则直接返回当前元素
  • 如果没有过期,也就是说剩余时间还存在,则先获取Leader对象,如果Leader已经有线程在处理,则当前线程进行无限期等待,如果Leader为空,则首先将Leader设置为当前线程,并且让当前线程等待剩余时间。
  • 最后将Leader线程设置为空
  1. 如果Leader已经为空,并且队列有内容则唤醒一个等待的队列。

poll操作

获取最早过期的元素,如果队列头没有过期的元素则直接返回null,反之返回过期的元素。

 1public E poll() {
2    final ReentrantLock lock = this.lock;
3    lock.lock();
4    try {
5        E first = q.peek();
6        // 如果队列为空或者队列最早过期的元素没有过期,则返回null。
7        if (first == null || first.getDelay(NANOSECONDS) > 0)
8            return null;
9        else
10            // 出队列操作。
11            return q.poll();
12    } finally {
13        lock.unlock();
14    }
15}

小结

  1. DelayQueue是一个无界的并发延迟阻塞队列,队列中的元素必须实现Delayed接口,相应了需要实现Comparable接口实现比较的方法
  2. Leader-Follower模式的变种,用于最小化不必要的定时等待,当一个线程被选择为Leader时,它会等待延迟过去执行代码逻辑,而其他线程则需要无限期等待,在从take或poll返回之前,每当队列的头部被替换为具有更早到期时间的元素时,leader字段将通过重置为空而无效,Leader线程必须向其中一个Follower线程发出信号,被唤醒的 follwer 线程被设置为新的Leader 线程。

喜欢的同学点赞关注下微信公众号,推送优质文章。

标签:队列,lock,元素,剖析,过期,线程,DelayQueue,Leader
来源: https://www.cnblogs.com/dwlsxj/p/delayedQueue.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有