纯净、安全、绿色的下载网站

首页|软件分类|下载排行|最新软件|IT学院

当前位置:首页IT学院IT技术

java DelayQueue延迟队列 java并发中DelayQueue延迟队列原理剖析

BattleHeart   2021-06-08 我要评论
想了解java并发中DelayQueue延迟队列原理剖析的相关内容吗BattleHeart在本文为您仔细讲解java DelayQueue延迟队列的相关知识和一些Code实例欢迎阅读和指正我们先划重点:java,DelayQueue延迟队列下面大家一起来学习吧

介绍

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

源码分析

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

 public interface Delayed extends Comparable<Delayed> {
 
     /**
      * Returns the remaining delay associated with this object, in the
      * given time unit.
      *
      * @param unit the time unit
      * @return the remaining delay; zero or negative values indicate
      * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

DelayQueue的成员变量如下所示:

 // 锁
 private final transient ReentrantLock lock = new ReentrantLock();
 // 优先队列
 private final PriorityQueue<E> q = new PriorityQueue<E>();
 
 /**
  * Leader-Follower的变种
  * Thread designated to wait for the element at the head of
  * the queue.  This variant of the Leader-Follower pattern
 * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
 * minimize unnecessary timed waiting.  When a thread becomes
 * the leader, it waits only for the next delay to elapse, but
 * other threads await indefinitely.  The leader thread must
 * signal some other thread before returning from take() or
 * poll(...), unless some other thread becomes leader in the
 * interim.  Whenever the head of the queue is replaced with
 * an element with an earlier expiration time, the leader
 * field is invalidated by being reset to null, and some
 * waiting thread, but not necessarily the current leader, is
 * signalled.  So waiting threads must be prepared to acquire
 * and lose leadership while waiting.
 */
private Thread leader = null;

/**
 * Condition signalled when a newer element becomes available
 * at the head of the queue or a new thread may need to
 * become leader.
 */
// 条件代表如果有数据则通知Follower线程唤醒线程处理队列内容
private final Condition available = lock.newCondition();

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

offer操作

 public boolean offer(E e) {
     // 获取到锁
     final ReentrantLock lock = this.lock;
     lock.lock();
     try {
         // 将元素存储到PriorityQueue优先队列中
         q.offer(e);
         // 如果第一个元素是当前元素说明之前队列中为空则先将Leader设置为空通知等待线程可以争抢Leader了
         if (q.peek() == e) {
            leader = null;
            available.signal();
        }
        // 返回成功
        return true;
    } finally {
        lock.unlock();
    }
}

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

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

put操作

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

public void put(E e) {
    offer(e);
}

take操作

 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);
                // 如果剩余过期时间小于0则说明已经过期反之还没有过期
                if (delay <= )
                    // 如果已经过期直接获取最早过期的元素并返回
                    return q.poll();
                // 如果剩余过期日期大于0则会进入到这里
                // 将刚才获取的最早过期的元素设置为空
                first = null; // don't retain ref while waiting
                // 如果有线程争抢的Leader线程则进行无限期等待
                if (leader != null)
                    // 无限期等待并让出锁
                    available.await();
                else {
                    // 获取当前线程
                    Thread thisThread = Thread.currentThread();
                    // 设置当前线程变为Leader线程
                    leader = thisThread;
                    try {
                        // 等待剩余等待时间
                        available.awaitNanos(delay);
                    } finally {
                        // 将Leader设置为null
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 如果队列不为空并且没有Leader则通知等待线程可以成为Leader
        if (leader == null && q.peek() != null)
            // 通知等待线程
            available.signal();
        lock.unlock();
    }
}
  1. 当获取元素时先获取到锁对象
  2. 获取最早过期的元素但是并不从队列中弹出元素
  3. 最早过期元素是否为空如果为空则直接让当前线程无限期等待状态并且让出当前锁对象
  4. 如果最早过期的元素不为空
    1. 获取最早过期元素的剩余过期时间如果已经过期则直接返回当前元素
    2. 如果没有过期也就是说剩余时间还存在则先获取Leader对象如果Leader已经有线程在处理则当前线程进行无限期等待如果Leader为空则首先将Leader设置为当前线程并且让当前线程等待剩余时间
    3. 最后将Leader线程设置为空
  5. 如果Leader已经为空并且队列有内容则唤醒一个等待的队列

poll操作

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

 public E poll() {
     final ReentrantLock lock = this.lock;
     lock.lock();
     try {
         E first = q.peek();
         // 如果队列为空或者队列最早过期的元素没有过期则返回null
         if (first == null || first.getDelay(NANOSECONDS) > 0)
             return null;
         else
            // 出队列操作
            return q.poll();
    } finally {
        lock.unlock();
    }
}

小结

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

相关文章

猜您喜欢

网友评论

Copyright 2020 www.fresh-weather.com 【世纪下载站】 版权所有 软件发布

声明:所有软件和文章来自软件开发商或者作者 如有异议 请与本站联系 点此查看联系方式