精华内容
参与话题
问答
  • DelayQueue

    千次阅读 2020-10-23 10:29:55
    DelayQueue 延迟队列底层使用的是锁的能力,比如说要在当前时间往后延迟 5 秒执行,那么当前线程就会沉睡 5 秒,等 5 秒后线程被唤醒时,如果能获取到资源的话,线程即可立马执行。原理上似乎很简单,但内部实现却很...

    引导语

    之前我们说的阻塞队列,都是资源足够时立马执行。本章我们说的队列比较特殊,是一种延迟队列,意思是延迟执行,并且可以设置延迟多久之后执行,比如设置过 5 秒钟之后再执行,在一些延迟执行的场景被大量使用,比如说延迟对账等等。

    1 整体设计

    DelayQueue 延迟队列底层使用的是锁的能力,比如说要在当前时间往后延迟 5 秒执行,那么当前线程就会沉睡 5 秒,等 5 秒后线程被唤醒时,如果能获取到资源的话,线程即可立马执行。原理上似乎很简单,但内部实现却很复杂,有很多难点,比如当运行资源不够,多个线程同时被唤醒时,如何排队等待?比如说在何时阻塞?何时开始执行等等?接下来我们从源码角度来看下是如何实现的。

    1.1 类注释

    类注释上比较简单,只说了三个概念:

    1. 队列中元素将在过期时被执行,越靠近队头,越早过期;
    2. 未过期的元素不能够被 take;
    3. 不允许空元素。

    这三个概念,其实就是三个问题,下文我们会一一看下这三点是如何实现的。

    1.2 类图

    DelayQueue 的类图和之前的队列一样,不多说,关键是 DelayQueue 类上是有泛型的,如下:

    public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E> {
    

    从泛型中可以看出,DelayQueue 中的元素必须是 Delayed 的子类,Delayed 是表达延迟能力的关键接口,其继承了 Comparable 接口,并定义了还剩多久过期的方法,如下:

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

    也就是说 DelayQueue 队列中的元素必须是实现 Delayed 接口和 Comparable 接口的,并覆写了 getDelay 方法和 compareTo 的方法才行,不然在编译时,编译器就会提醒我们元素必须强制实现 Delayed 接口。

    除此之外 DelayQueue 还大量使用了 PriorityQueue 队列的大量功能,这个和 SynchronousQueue 队列很像,大量复用了其它基础类的逻辑,代码示例如下:

    link

    PriorityQueue 中文叫做优先级队列,在此处的作用就是可以根据过期时间做优先级排序,让先过期的可以先执行,用来实现类注释中的第一点。

    这里的复用的思想还是蛮重要的,我们在源码中经常会遇到这种思想,比如说 LinkedHashMap 复用 HashMap 的能力,Set 复用 Map 的能力,还有此处的 DelayQueue 复用 PriorityQueue 的能力。小结一下,如果想要复用需要做到哪些:

    需要把能遇见可复用的功能尽量抽象,并开放出可扩展的地方,比如说 HashMap 在操作数组的方法中,都给 LinkedHashMap 开放出很多 after 开头的方法,便于 LinkedHashMap 进行排序、删除等等;
    采用组合或继承两种手段进行复用,比如 LinkedHashMap 采用的继承、 Set 和 DelayQueue 采用的组合,组合的意思就是把可复用的类给依赖进来。

    2 演示

    为了方便大家理解,写了一个演示的 demo,演示了一下:

    public class DelayQueueDemo {
        // 队列消息的生产者
      static class Product implements Runnable {
        private final BlockingQueue queue;
        public Product(BlockingQueue queue) {
          this.queue = queue;
        }
        
        @Override
        public void run() {
          try {
            log.info("begin put");
            long beginTime = System.currentTimeMillis();
            queue.put(new DelayedDTO(System.currentTimeMillis() + 2000L,beginTime));//延迟 2 秒执行
            queue.put(new DelayedDTO(System.currentTimeMillis() + 5000L,beginTime));//延迟 5 秒执行
            queue.put(new DelayedDTO(System.currentTimeMillis() + 1000L * 10,beginTime));//延迟 10 秒执行
            log.info("end put");
          } catch (InterruptedException e) {
            log.error("" + e);
          }
        }
      }
        // 队列的消费者
      static class Consumer implements Runnable {
        private final BlockingQueue queue;
        public Consumer(BlockingQueue queue) {
          this.queue = queue;
        }
     
        @Override
        public void run() {
          try {
            log.info("Consumer begin");
            ((DelayedDTO) queue.take()).run();
            ((DelayedDTO) queue.take()).run();
            ((DelayedDTO) queue.take()).run();
            log.info("Consumer end");
          } catch (InterruptedException e) {
            log.error("" + e);
          }
        }
      }
     
      @Data
      // 队列元素,实现了 Delayed 接口
      static class DelayedDTO implements Delayed {
        Long s;
        Long beginTime;
        public DelayedDTO(Long s,Long beginTime) {
          this.s = s;
          this.beginTime =beginTime;
        }
     
        @Override
        public long getDelay(TimeUnit unit) {
          return unit.convert(s - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
     
        @Override
        public int compareTo(Delayed o) {
          return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }
     
        public void run(){
          log.info("现在已经过了{}秒钟",(System.currentTimeMillis() - beginTime)/1000);
        }
      }
        // demo 运行入口
      public static void main(String[] args) throws InterruptedException {
        BlockingQueue q = new DelayQueue();
        DelayQueueDemo.Product p = new DelayQueueDemo.Product(q);
        DelayQueueDemo.Consumer c = new DelayQueueDemo.Consumer(q);
        new Thread(c).start();
        new Thread(p).start();
      }
    }
    

    打印出来的结果如下:

    06:57:50.544 [Thread-0] Consumer begin
    06:57:50.544 [Thread-1] begin put
    06:57:50.551 [Thread-1] end put
    06:57:52.554 [Thread-0] 延迟了2秒钟才执行
    06:57:55.555 [Thread-0] 延迟了5秒钟才执行
    06:58:00.555 [Thread-0] 延迟了10秒钟才执行
    06:58:00.556 [Thread-0] Consumer end
    

    写这个代码的目的主要想演示一下延迟执行的例子,我们大概的思路是:

    1. 新建队列的元素,如 DelayedDTO,必须实现 Delayed 接口,我们在 getDelay 方法中实现了现在离过期时间还剩多久的方法。
    2. 定义队列元素的生产者,和消费者,对应着代码中的 Product 和 Consumer。
    3. 对生产者和消费者就行初始化和管理,对应着我们的 main 方法。

    虽然这只是一个简单的 demo,但实际工作中,我们使用 DelayQueue 基本上就是这种思想,只不过写代码的时候会更加通用和周全,接下来我们来看下 DelayQueue 是如何实现 put 和 take 的。

    3 放数据

    我们以 put 为例,put 调用的是 offer 的方法,offer 的源码如下:

    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        // 上锁
        lock.lock();
        try {
            // 使用 PriorityQueue 的扩容,排序等能力
            q.offer(e);
            // 如果恰好刚放进去的元素正好在队列头
            // 立马唤醒 take 的阻塞线程,执行 take 操作
            // 如果元素需要延迟执行的话,可以使其更快的沉睡计时
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            // 释放锁
            lock.unlock();
        }
    }
    

    可以看到其实底层使用到的是 PriorityQueue 的 offer 方法,我们来看下:

    // 新增元素
    public boolean offer(E e) {
        // 如果是空元素的话,抛异常
        if (e == null)
            throw new NullPointerException();
        modCount++;
        int i = size;
        // 队列实际大小大于容量时,进行扩容
        // 扩容策略是:如果老容量小于 64,2 倍扩容,如果大于 64,50 % 扩容
        if (i >= queue.length)
            grow(i + 1);
        size = i + 1;
        // 如果队列为空,当前元素正好处于队头
        if (i == 0)
            queue[0] = e;
        else
        // 如果队列不为空,需要根据优先级进行排序
            siftUp(i, e);
        return true;
    }
    // 按照从小到大的顺序排列
        private void siftUpComparable(int k, E x) {
            Comparable<? super E> key = (Comparable<? super E>) x;
            // k 是当前队列实际大小的位置
            while (k > 0) {
                // 对 k 进行减倍
                int parent = (k - 1) >>> 1;
                Object e = queue[parent];
                // 如果 x 比 e 大,退出,把 x 放在 k 位置上
                if (key.compareTo((E) e) >= 0)
                    break;
                // x 比 e 小,继续循环,直到找到 x 比队列中元素大的位置
                queue[k] = e;
                k = parent;
            }
            queue[k] = key;
        }
    

    可以看到,PriorityQueue 的 offer 方法主要做了三件事情:

    1. 对新增元素进行判空;
    2. 对队列进行扩容,扩容策略和集合的扩容策略很相近;
    3. 根据元素的 compareTo 方法进行排序,我们希望最终排序的结果是从小到大的,因为我们想让队头的都是过期的数据,我们需要在 compareTo 方法里面实现:通过每个元素的过期时间进行排序,如下:
    (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    

    这样便可实现越快过期的元素越能排到队头。

    可以看到,新增数据时,只是使用到了 compareTo 方法,来对队列中的元素进行排序,接下来我们看下,取数据时,是如何操作的。

    4 拿数据

    取数据时,如果发现有元素的过期时间到了,就能拿出数据来,如果没有过期元素,那么线程就会一直阻塞,我们以 take 为例子,来看一下核心源码:

    for (;;) {
        // 从队头中拿数据出来
        E first = q.peek();
        // 如果为空,说明队列中,没有数据,阻塞住
        if (first == null)
            available.await();
        else {
            // 获取队头数据的过期时间
            long delay = first.getDelay(NANOSECONDS);
            // 如果过期了,直接返回队头数据
            if (delay <= 0)
                return q.poll();
            // 引用置为 null ,便于 gc,这样可以让线程等待时,回收 first 变量
            first = null;
            // leader 不为空的话,表示当前队列元素之前已经被设置过阻塞时间了
            // 直接阻塞当前线程等待。
            if (leader != null)
                available.await();
            else {
              // 之前没有设置过阻塞时间,按照一定的时间进行阻塞
                Thread thisThread = Thread.currentThread();
                leader = thisThread;
                try {
                    // 进行阻塞
                    available.awaitNanos(delay);
                } finally {
                    if (leader == thisThread)
                        leader = null;
                }
            }
        }
    }
    

    可以看到阻塞等待的功能底层使用的是锁的能力,这个我们在后面章节中会说到。

    以上演示的 take 方法是会无限阻塞,直到队头的过期时间到了才会返回,如果不想无限阻塞,可以尝试 poll 方法,设置超时时间,在超时时间内,队头元素还没有过期的话,就会返回 null。

    5 总结

    DelayQueue 是非常有意思的队列,底层使用了排序和超时阻塞实现了延迟队列,排序使用的是 PriorityQueue 排序能力,超时阻塞使用得是锁的等待能力,可以看出 DelayQueue 其实就是为了满足延迟执行的场景,在已有 API 的基础上进行了封装,我们在工作中,可以学习这种思想,对已有的功能能复用的尽量复用,减少开发的工作量。

    展开全文

空空如也

1 2 3 4 5 ... 20
收藏数 1,389
精华内容 555
关键字:

delayqueue