阻塞队列—DelayedWorkQueue源码分析

 前言

颍上ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为创新互联的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：028-86922220（备注：SSL证书合作）期待与您的合作！

线程池运行时，会不断从任务队列中获取任务，然后执行任务。如果我们想实现延时或者定时执行任务，重要一点就是任务队列会根据任务延时时间的不同进行排序，延时时间越短地就排在队列的前面，先被获取执行。

队列是先进先出的数据结构，就是先进入队列的数据，先被获取。但是有一种特殊的队列叫做优先级队列，它会对插入的数据进行优先级排序，保证优先级越高的数据首先被获取，与数据的插入顺序无关。

实现优先级队列高效常用的一种方式就是使用堆。关于堆的实现可以查看《堆和二叉堆的实现和特性》

ScheduledThreadPoolExecutor线程池

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，所以其内部的数据结构和ThreadPoolExecutor基本一样，并在其基础上增加了按时间调度执行任务的功能，分为延迟执行任务和周期性执行任务。

ScheduledThreadPoolExecutor的构造函数只能传3个参数corePoolSize、ThreadFactory、RejectedExecutionHandler，默认maximumPoolSize为Integer.MAX_VALUE。

工作队列是高度定制化的延迟阻塞队列DelayedWorkQueue，其实现原理和DelayQueue基本一样，核心数据结构是二叉最小堆的优先队列，队列满时会自动扩容，所以offer操作永远不会阻塞，maximumPoolSize也就用不上了，所以线程池中永远会保持至多有corePoolSize个工作线程正在运行。

 
 
 
 
    
  
  
  
  
  1. public ScheduledThreadPoolExecutor(int corePoolSize,
    2. 
  
  
  
  
  2.                                    ThreadFactory threadFactory,
    3. 
  
  
  
  
  3.                                    RejectedExecutionHandler handler) {
    4. 
  
  
  
  
  4.     super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
    5. 
  
  
  
  
  5.           new DelayedWorkQueue(), threadFactory, handler);
    6. 
  
  
  
  
  6. }
    7.

 DelayedWorkQueue延迟阻塞队列

DelayedWorkQueue 也是一种设计为定时任务的延迟队列，它的实现和DelayQueue一样，不过是将优先级队列和DelayQueue的实现过程迁移到本身方法体中，从而可以在该过程当中灵活的加入定时任务特有的方法调用。

工作原理

DelayedWorkQueue的实现原理中规中矩，内部维护了一个以RunnableScheduledFuture类型数组实现的最小二叉堆，初始容量是16，使用ReentrantLock和Condition实现生产者和消费者模式。

源码分析

定义

DelayedWorkQueue 的类继承关系如下：

其包含的方法定义如下：

成员属性

 
 
 
 
    
  
  
  
  
  1. // 初始时，数组长度大小。
    2. 
  
  
  
  
  2. private static final int INITIAL_CAPACITY = 16;        
    3. 
  
  
  
  
  3. // 使用数组来储存队列中的元素。
    4. 
  
  
  
  
  4. private RunnableScheduledFuture[] queue =  new RunnableScheduledFuture[INITIAL_CAPACITY];        
    5. 
  
  
  
  
  5. // 使用lock来保证多线程并发安全问题。
    6. 
  
  
  
  
  6. private final ReentrantLock lock = new ReentrantLock();        
    7. 
  
  
  
  
  7. // 队列中储存元素的大小
    8. 
  
  
  
  
  8. private int size = 0;        
    9. 
  
  
  
  
  9. //特指队列头任务所在线程
    10. 
  
  
  
  
  10. private Thread leader = null;        
    11. 
  
  
  
  
  11. // 当队列头的任务延时时间到了，或者有新的任务变成队列头时，用来唤醒等待线程
    12. 
  
  
  
  
  12. private final Condition available = lock.newCondition();
    13.

 DelayedWorkQueue是用数组来储存队列中的元素，核心数据结构是二叉最小堆的优先队列，队列满时会自动扩容。

构造函数

DelayedWorkQueue 是 ScheduledThreadPoolExecutor 的静态类部类，默认只有一个无参构造方法。

 
 
 
 
    
  
  
  
  
  1. static class DelayedWorkQueue extends AbstractQueue
    2. 
  
  
  
  
  2.         implements BlockingQueue {
    3. 
  
  
  
  
  3.  // ...
    4. 
  
  
  
  
  4. }
    5.

 入队方法

DelayedWorkQueue 提供了 put/add/offer(带时间) 三个插入元素方法。我们发现与普通阻塞队列相比，这三个添加方法都是调用offer方法。那是因为它没有队列已满的条件，也就是说可以不断地向DelayedWorkQueue添加元素,当元素个数超过数组长度时，会进行数组扩容。

 
 
 
 
    
  
  
  
  
  1. public void put(Runnable e) {
    2. 
  
  
  
  
  2.  offer(e);
    3. 
  
  
  
  
  3. }        
    4. 
  
  
  
  
  4. public boolean add(Runnable e) {            
    5. 
  
  
  
  
  5.     return offer(e);
    6. 
  
  
  
  
  6. }        
    7. 
  
  
  
  
  7. public boolean offer(Runnable e, long timeout, TimeUnit unit) {            
    8. 
  
  
  
  
  8.     return offer(e);
    9. 
  
  
  
  
  9. }
    10.

 offer添加元素

ScheduledThreadPoolExecutor提交任务时调用的是DelayedWorkQueue.add，而add、put等一些对外提供的添加元素的方法都调用了offer。

 
 
 
 
    
  
  
  
  
  1. public boolean offer(Runnable x) {            
    2. 
  
  
  
  
  2.     if (x == null)                
    3. 
  
  
  
  
  3.         throw new NullPointerException();
    4. 
  
  
  
  
  4.     RunnableScheduledFuture e = (RunnableScheduledFuture)x;            
    5. 
  
  
  
  
  5.     // 使用lock保证并发操作安全
    6. 
  
  
  
  
  6.     final ReentrantLock lock = this.lock;
    7. 
  
  
  
  
  7.     lock.lock();            
    8. 
  
  
  
  
  8.     try {                
    9. 
  
  
  
  
  9.         int i = size;                
    10. 
  
  
  
  
  10.         // 如果要超过数组长度，就要进行数组扩容
    11. 
  
  
  
  
  11.         if (i >= queue.length)                    
    12. 
  
  
  
  
  12.             // 数组扩容
    13. 
  
  
  
  
  13.             grow();                
    14. 
  
  
  
  
  14.         // 将队列中元素个数加一
    15. 
  
  
  
  
  15.         size = i + 1;                
    16. 
  
  
  
  
  16.         // 如果是第一个元素，那么就不需要排序，直接赋值就行了
    17. 
  
  
  
  
  17.         if (i == 0) {
    18. 
  
  
  
  
  18.          queue[0] = e;
    19. 
  
  
  
  
  19.             setIndex(e, 0);
    20. 
  
  
  
  
  20.         } else {                    
    21. 
  
  
  
  
  21.             // 调用siftUp方法，使插入的元素变得有序。
    22. 
  
  
  
  
  22.             siftUp(i, e);
    23. 
  
  
  
  
  23.         }                
    24. 
  
  
  
  
  24.         // 表示新插入的元素是队列头，更换了队列头，
    25. 
  
  
  
  
  25.         // 那么就要唤醒正在等待获取任务的线程。
    26. 
  
  
  
  
  26.         if (queue[0] == e) {
    27. 
  
  
  
  
  27.          leader = null;                    
    28. 
  
  
  
  
  28.             // 唤醒正在等待等待获取任务的线程
    29. 
  
  
  
  
  29.             available.signal();
    30. 
  
  
  
  
  30.         }
    31. 
  
  
  
  
  31.     } finally {
    32. 
  
  
  
  
  32.      lock.unlock();
    33. 
  
  
  
  
  33.     }            
    34. 
  
  
  
  
  34.     return true;
    35. 
  
  
  
  
  35. }
    36.

 其基本流程如下：

  1. 其作为生产者的入口，首先获取锁。
  2. 判断队列是否要满了(size >= queue.length)，满了就扩容grow()。
  3. 队列未满，size+1。
  4. 判断添加的元素是否是第一个，是则不需要堆化。
  5. 添加的元素不是第一个，则需要堆化siftUp。
  6. 如果堆顶元素刚好是此时被添加的元素，则唤醒take线程消费。
  7. 最终释放锁。

offer基本流程图如下：

扩容grow()

可以看到，当队列满时，不会阻塞等待，而是继续扩容。新容量newCapacity在旧容量oldCapacity的基础上扩容50%(oldCapacity >> 1相当于oldCapacity /2)。最后Arrays.copyOf，先根据newCapacity创建一个新的空数组，然后将旧数组的数据复制到新数组中。

 
 
 
 
    
  
  
  
  
  1. private void grow() {            
    2. 
  
  
  
  
  2.     int oldCapacity = queue.length;            
    3. 
  
  
  
  
  3.     // 每次扩容增加原来数组的一半数量。
    4. 
  
  
  
  
  4.     // grow 50%
    5. 
  
  
  
  
  5.     int newCapacity = oldCapacity + (oldCapacity >> 1); 
    6. 
  
  
  
  
  6.     if (newCapacity < 0) // overflow
    7. 
  
  
  
  
  7.      newCapacity = Integer.MAX_VALUE;            
    8. 
  
  
  
  
  8.     // 使用Arrays.copyOf来复制一个新数组
    9. 
  
  
  
  
  9.     queue = Arrays.copyOf(queue, newCapacity);
    10. 
  
  
  
  
  10. }
    11.

 向上堆化siftUp

新添加的元素先会加到堆底，然后一步步和上面的父亲节点比较，若小于父亲节点则和父亲节点互换位置，循环比较直至大于父亲节点才结束循环。通过循环，来查找元素key应该插入在堆二叉树那个节点位置，并交互父节点的位置。

向上堆化siftUp的详细过程可以查看《堆和二叉堆的实现和特性》

 
 
 
 
    
  
  
  
  
  1. private void siftUp(int k, RunnableScheduledFuture key) {            
    2. 
  
  
  
  
  2.     // 当k==0时，就到了堆二叉树的根节点了，跳出循环
    3. 
  
  
  
  
  3.     while (k > 0) {                
    4. 
  
  
  
  
  4.         // 父节点位置坐标, 相当于(k - 1) / 2
    5. 
  
  
  
  
  5.         int parent = (k - 1) >>> 1;                
    6. 
  
  
  
  
  6.         // 获取父节点位置元素
    7. 
  
  
  
  
  7.         RunnableScheduledFuture e = queue[parent];                
    8. 
  
  
  
  
  8.         // 如果key元素大于父节点位置元素，满足条件，那么跳出循环
    9. 
  
  
  
  
  9.         // 因为是从小到大排序的。
    10. 
  
  
  
  
  10.         if (key.compareTo(e) >= 0)                    
    11. 
  
  
  
  
  11.             break;                
    12. 
  
  
  
  
  12.         // 否则就将父节点元素存放到k位置
    13. 
  
  
  
  
  13.         queue[k] = e;                
    14. 
  
  
  
  
  14.         // 这个只有当元素是ScheduledFutureTask对象实例才有用，用来快速取消任务。
    15. 
  
  
  
  
  15.         setIndex(e, k);                
    16. 
  
  
  
  
  16.         // 重新赋值k，寻找元素key应该插入到堆二叉树的那个节点
    17. 
  
  
  
  
  17.         k = parent;
    18. 
  
  
  
  
  18.     }            
    19. 
  
  
  
  
  19.     // 循环结束，k就是元素key应该插入的节点位置
    20. 
  
  
  
  
  20.     queue[k] = key;
    21. 
  
  
  
  
  21.     setIndex(key, k);
    22. 
  
  
  
  
  22. }
    23.

 出队方法

DelayedWorkQueue 提供了以下几个出队方法

  • take()，等待获取队列头元素
  • poll() ，立即获取队列头元素
  • poll(long timeout, TimeUnit unit) ，超时等待获取队列头元素

take消费元素

Worker工作线程启动后就会循环消费工作队列中的元素，因为ScheduledThreadPoolExecutor的keepAliveTime=0，所以消费任务其只调用了DelayedWorkQueue.take。take基本流程如下：

  • 首先获取可中断锁，判断堆顶元素是否是空，空的则阻塞等待available.await()。
  • 堆顶元素不为空，则获取其延迟执行时间delay，delay <= 0说明到了执行时间，出队列finishPoll。
  • delay > 0还没到执行时间，判断leader线程是否为空，不为空则说明有其他take线程也在等待，当前take将无限期阻塞等待。
  • leader线程为空，当前take线程设置为leader，并阻塞等待delay时长。
  • 当前leader线程等待delay时长自动唤醒或者被其他take线程唤醒，则最终将leader设置为null。
  • 再循环一次判断delay <= 0出队列。
  • 跳出循环后判断leader为空并且堆顶元素不为空，则唤醒其他take线程，最后是否锁。
 
 
 
 
    
  
  
  
  
  1. public RunnableScheduledFuture take() throws InterruptedException {            
    2. 
  
  
  
  
  2.     final ReentrantLock lock = this.lock;
    3. 
  
  
  
  
  3.     lock.lockInterruptibly();            
    4. 
  
  
  
  
  4.     try {                
    5. 
  
  
  
  
  5.         for (;;) {
    6. 
  
  
  
  
  6.          RunnableScheduledFuture first = queue[0];                    
    7. 
  
  
  
  
  7.             // 如果没有任务，就让线程在available条件下等待。
    8. 
  
  
  
  
  8.             if (first == null)
    9. 
  
  
  
  
  9.              available.await();                    
    10. 
  
  
  
  
  10.             else {                        
    11. 
  
  
  
  
  11.                 // 获取任务的剩余延时时间
    12. 
  
  
  
  
  12.                 long delay = first.getDelay(NANOSECONDS);                        
    13. 
  
  
  
  
  13.                 // 如果延时时间到了，就返回这个任务，用来执行。
    14. 
  
  
  
  
  14.                 if (delay <= 0)                            
    15. 
  
  
  
  
  15.                     return finishPoll(first);                        
    16. 
  
  
  
  
  16.                 // 将first设置为null，当线程等待时，不持有first的引用
    17. 
  
  
  
  
  17.                 first = null; // don't retain ref while waiting
    18. 
  
  
  
  
  18. 
  
  
  
  
  19.                 // 如果还是原来那个等待队列头任务的线程，
    20. 
  
  
  
  
  20.                 // 说明队列头任务的延时时间还没有到，继续等待。
    21. 
  
  
  
  
  21.                 if (leader != null)
    22. 
  
  
  
  
  22.                  available.await();                        
    23. 
  
  
  
  
  23.                 else {                            
    24. 
  
  
  
  
  24.                     // 记录一下当前等待队列头任务的线程
    25. 
  
  
  
  
  25.                     Thread thisThread = Thread.currentThread();
    26. 
  
  
  
  
  26.                     leader = thisThread;                            
    27. 
  
  
  
  
  27.                     try {                                
    28. 
  
  
  
  
  28.                         // 当任务的延时时间到了时，能够自动超时唤醒。
    29. 
  
  
  
  
  29.                         available.awaitNanos(delay);
    30. 
  
  
  
  
  30.                     } finally {                                
    31. 
  
  
  
  
  31.                         if (leader == thisThread)
    32. 
  
  
  
  
  32.                          leader = null;
    33. 
  
  
  
  
  33.                     }
    34. 
  
  
  
  
  34.                 }
    35. 
  
  
  
  
  35.             }
    36. 
  
  
  
  
  36.         }
    37. 
  
  
  
  
  37.     } finally {                
    38. 
  
  
  
  
  38.         if (leader == null && queue[0] != null)                    // 唤醒等待任务的线程
    39. 
  
  
  
  
  39.          available.signal();
    40. 
  
  
  
  
  40.         ock.unlock();
    41. 
  
  
  
  
  41.     }
    42. 
  
  
  
  
  42. }
    43.

 take基本流程图如下：

take线程阻塞等待

可以看出这个生产者take线程会在两种情况下阻塞等待：

  • 堆顶元素为空。
  • 堆顶元素的delay > 0 。

finishPoll出队列

堆顶元素delay<=0，执行时间到，出队列就是一个向下堆化的过程siftDown。

 
 
 
 
    
  
  
  
  
  1. // 移除队列头元素
    2. 
  
  
  
  
  2. private RunnableScheduledFuture finishPoll(RunnableScheduledFuture f) {            
    3. 
  
  
  
  
  3.     // 将队列中元素个数减一
    4. 
  
  
  
  
  4.     int s = --size;            
    5. 
  
  
  
  
  5.     // 获取队列末尾元素x
    6. 
  
  
  
  
  6.     RunnableScheduledFuture x = queue[s];            
    7. 
  
  
  
  
  7.     // 原队列末尾元素设置为null
    8. 
  
  
  
  
  8.     queue[s] = null;            
    9. 
  
  
  
  
  9.     if (s != 0)                
    10. 
  
  
  
  
  10.         // 因为移除了队列头元素，所以进行重新排序。
    11. 
  
  
  
  
  11.         siftDown(0, x);
    12. 
  
  
  
  
  12.     setIndex(f, -1);            
    13. 
  
  
  
  
  13.     return f;
    14. 
  
  
  
  
  14. }
    15.

堆的删除方法主要分为三步：

  1. 先将队列中元素个数减一;
  2. 将原队列末尾元素设置成为队列头元素，再将队列末尾元素设置为null;
  3. 调用setDown(O,x)方法，保证按照元素的优先级排序。

向下堆化siftDown

由于堆顶元素出队列后，就破坏了堆的结构，需要组织整理下，将堆尾元素移到堆顶，然后向下堆化：

  • 从堆顶开始，父亲节点与左右子节点中较小的孩子节点比较(左孩子不一定小于右孩子)。
  • 父亲节点小于等于较小孩子节点，则结束循环，不需要交换位置。
  • 若父亲节点大于较小孩子节点，则交换位置。
  • 继续向下循环判断父亲节点和孩子节点的关系，直到父亲节点小于等于较小孩子节点才结束循环。

向下堆化siftDown的详细过程可以查看《堆和二叉堆的实现和特性》

 
 
 
 
    
  
  
  
  
  1. private void siftDown(int k, RunnableScheduledFuture key) {     
    2. 
  
  
  
  
  2.     // 无符号右移，相当于size/2
    3. 
  
  
  
  
  3.     int half = size >>> 1;            
    4. 
  
  
  
  
  4.     // 通过循环，保证父节点的值不能大于子节点。
    5. 
  
  
  
  
  5.     while (k < half) {                
    6. 
  
  
  
  
  6.         // 左子节点, 相当于 (k * 2) + 1
    7. 
  
  
  
  
  7.         int child = (k << 1) + 1;                
    8. 
  
  
  
  
  8.         // 左子节点位置元素
    9. 
  
  
  
  
  9.         RunnableScheduledFuture c = queue[child];                
    10. 
  
  
  
  
  10.         // 右子节点, 相当于 (k * 2) + 2
    11. 
  
  
  
  
  11.         int right = child + 1;                
    12. 
  
  
  
  
  12.         // 如果左子节点元素值大于右子节点元素值，那么右子节点才是较小值的子节点。
    13. 
  
  
  
  
  13.         // 就要将c与child值重新赋值
    14. 
  
  
  
  
  14.         if (right < size && c.compareTo(queue[right]) > 0)
    15. 
  
  
  
  
  15.          c = queue[child = right];                
    16. 
  
  
  
  
  16.         // 如果父节点元素值小于较小的子节点元素值，那么就跳出循环
    17. 
  
  
  
  
  17.         if (key.compareTo(c) <= 0)                    
    18. 
  
  
  
  
  18.             break;                
    19. 
  
  
  
  
  19.         // 否则，父节点元素就要和子节点进行交换
    20. 
  
  
  
  
  20.         queue[k] = c;
    21. 
  
  
  
  
  21.         setIndex(c, k);
    22. 
  
  
  
  
  22.         k = child;
    23. 
  
  
  
  
  23.     }            
    24. 
  
  
  
  
  24.     queue[k] = key;
    25. 
  
  
  
  
  25.     setIndex(key, k);
    26. 
  
  
  
  
  26. }
    27.

 leader线程

leader线程的设计，是Leader-Follower模式的变种，旨在于为了不必要的时间等待。当一个take线程变成leader线程时，只需要等待下一次的延迟时间，而不是leader线程的其他take线程则需要等leader线程出队列了才唤醒其他take线程。

poll()

立即获取队列头元素，当队列头任务是null，或者任务延时时间没有到，表示这个任务还不能返回，因此直接返回null。否则调用finishPoll方法，移除队列头元素并返回。

 
 
 
 
    
  
  
  
  
  1. public RunnableScheduledFuture poll() {            
    2. 
  
  
  
  
  2.     final ReentrantLock lock = this.lock;
    3. 
  
  
  
  
  3.     lock.lock();            
    4. 
  
  
  
  
  4.     try {
    5. 
  
  
  
  
  5.      RunnableScheduledFuture first = queue[0];                
    6. 
  
  
  
  
  6.         // 队列头任务是null，或者任务延时时间没有到，都返回null
    7. 
  
  
  
  
  7.         if (first == null || first.getDelay(NANOSECONDS) > 0)                    
    8. 
  
  
  
  
  8.             return null;                
    9. 
  
  
  
  
  9.         else
    10. 
  
  
  
  
  10.          // 移除队列头元素
    11. 
  
  
  
  
  11.             return finishPoll(first);
    12. 
  
  
  
  
  12.     } finally {
    13. 
  
  
  
  
  13.      lock.unlock();
    14. 
  
  
  
  
  14.     }
    15. 
  
  
  
  
  15. }
    16.

 poll(long timeout, TimeUnit unit)

超时等待获取队列头元素，与take方法相比较，就要考虑设置的超时时间，如果超时时间到了，还没有获取到有用任务，那么就返回null。其他的与take方法中逻辑一样。

 
 
 
 
    
  
  
  
  
  1. public RunnableScheduledFuture poll(long timeout, TimeUnit unit)            
    2. 
  
  
  
  
  2.     throws InterruptedException {            
    3. 
  
  
  
  
  3.     long nanos = unit.toNanos(timeout);            
    4. 
  
  
  
  
  4.     final ReentrantLock lock = this.lock;
    5. 
  
  
  
  
  5.     lock.lockInterruptibly();            
    6. 
  
  
  
  
  6.     try {                
    7. 
  
  
  
  
  7.         for (;;) {
    8. 
  
  
  
  
  8.          RunnableScheduledFuture first = queue[0];                    
    9. 
  
  
  
  
  9.             // 如果没有任务。
    10. 
  
  
  
  
  10.             if (first == null) {                        
    11. 
  
  
  
  
  11.              // 超时时间已到，那么就直接返回null
    12. 
  
  
  
  
  12.                 if (nanos <= 0)                            
    13. 
  
  
  
  
  13.                     return null;                        
    14. 
  
  
  
  
  14.                 else
    15. 
  
  
  
  
  15.                  // 否则就让线程在available条件下等待nanos时间
    16. 
  
  
  
  
  16.                     nanos = available.awaitNanos(nanos);
    17. 
  
  
  
  
  17.             } else {                        
    18. 
  
  
  
  
  18.                 // 获取任务的剩余延时时间
    19. 
  
  
  
  
  19.                 long delay = first.getDelay(NANOSECONDS);                        
    20. 
  
  
  
  
  20.                 // 如果延时时间到了，就返回这个任务，用来执行。
    21. 
  
  
  
  
  21.                 if (delay <= 0)                            
    22. 
  
  
  
  
  22.                     return finishPoll(first);                        
    23. 
  
  
  
  
  23.                 // 如果超时时间已到，那么就直接返回null
    24. 
  
  
  
  
  24.                 if (nanos <= 0)                            
    25. 
  
  
  
  
  25.                     return null;                        
    26. 
  
  
  
  
  26.                 // 将first设置为null，当线程等待时，不持有first的引用
    27. 
  
  
  
  
  27.                 first = null; // don't retain ref while waiting
    28. 
  
  
  
  
  28.                 // 如果超时时间小于任务的剩余延时时间，那么就有可能获取不到任务。
    29. 
  
  
  
  
  29.                 // 在这里让线程等待超时时间nanos
    30. 
  
  
  
  
  30.                 if (nanos < delay || leader != null)
    31. 
  
  
  
  
  31.                  nanos = available.awaitNanos(nanos);                        
    32. 
  
  
  
  
  32.                 else {
    33. 
  
  
  
  
  33.                     Thread thisThread = Thread.currentThread();
    34. 
  
  
  
  
  34.                     leader = thisThread;                            
    35. 
  
  
  
  
  35.                     try {                                
    36. 
  
  
  
  
  36.                         // 当任务的延时时间到了时，能够自动超时唤醒。
    37. 
  
  
  
  
  37.                         long timeLeft = available.awaitNanos(delay);                                
    38. 
  
  
  
  
  38.                         // 计算剩余的超时时间
    39. 
  
  
  
  
  39.                         nanos -= delay - timeLeft;
    40. 
  
  
  
  
  40.                     } finally {                                
    41. 
  
  
  
  
  41.                         if (leader == thisThread)
    42. 
  
  
  
  
  42.                          leader = null;
    43. 
  
  
  
  
  43.                     }
    44. 
  
  
  
  
  44.                 }
    45. 
  
  
  
  
  45.             }
    46. 
  
  
  
  
  46.         }
    47. 
  
  
  
  
  47.     } finally {                
    48. 
  
  
  
  
  48.         if (leader == null && queue[0] != null)                    // 唤醒等待任务的线程
    49. 
  
  
  
  
  49.          available.signal();
    50. 
  
  
  
  
  50.         lock.unlock();
    51. 
  
  
  
  
  51.     }
    52. 
  
  
  
  
  52. }
    53.

 remove删除指定元素

删除指定元素一般用于取消任务时，任务还在阻塞队列中，则需要将其删除。当删除的元素不是堆尾元素时，需要做堆化处理。

 
 
 
 
    
  
  
  
  
  1. public boolean remove(Object x) {
    2. 
  
  
  
  
  2.     final ReentrantLock lock = this.lock;
    3. 
  
  
  
  
  3.     lock.lock();
    4. 
  
  
  
  
  4.     try {
    5. 
  
  
  
  
  5.         int i = indexOf(x);
    6. 
  
  
  
  
  6.         if (i < 0)
    7. 
  
  
  
  
  7.             return false;
    8. 
  
  
  
  
  8.         //维护heapIndex
    9. 
  
  
  
  
  9.         setIndex(queue[i], -1);
    10. 
  
  
  
  
  10.         int s = --size;
    11. 
  
  
  
  
  11.         RunnableScheduledFuture replacement = queue[s];
    12. 
  
  
  
  
  12.         queue[s] = null;
    13. 
  
  
  
  
  13.         if (s != i) {
    14. 
  
  
  
  
  14.             //删除的不是堆尾元素，则需要堆化处理
    15. 
  
  
  
  
  15.             //先向下堆化
    16. 
  
  
  
  
  16.             siftDown(i, replacement);
    17. 
  
  
  
  
  17.             if (queue[i] == replacement)
    18. 
  
  
  
  
  18.                 //若向下堆化后，i位置的元素还是replacement，说明四无需向下堆化的，
    19. 
  
  
  
  
  19.                 //则需要向上堆化
    20. 
  
  
  
  
  20.                 siftUp(i, replacement);
    21. 
  
  
  
  
  21.         }
    22. 
  
  
  
  
  22.         return true;
    23. 
  
  
  
  
  23.     } finally {
    24. 
  
  
  
  
  24.         lock.unlock();
    25. 
  
  
  
  
  25.     }
    26. 
  
  
  
  
  26. }
    27.

 总结

使用优先级队列DelayedWorkQueue，保证添加到队列中的任务，会按照任务的延时时间进行排序，延时时间少的任务首先被获取。

  1. DelayedWorkQueue的数据结构是基于堆实现的;
  2. DelayedWorkQueue采用数组实现堆，根节点出队，用最后叶子节点替换，然后下推至满足堆成立条件;最后叶子节点入队，然后向上推至满足堆成立条件;
  3. DelayedWorkQueue添加元素满了之后会自动扩容原来容量的1/2，即永远不会阻塞，最大扩容可达Integer.MAX_VALUE，所以线程池中至多有corePoolSize个工作线程正在运行;
  4. DelayedWorkQueue 消费元素take，在堆顶元素为空和delay >0 时，阻塞等待;
  5. DelayedWorkQueue 是一个生产永远不会阻塞，消费可以阻塞的生产者消费者模式;
  6. DelayedWorkQueue 有一个leader线程的变量，是Leader-Follower模式的变种。当一个take线程变成leader线程时，只需要等待下一次的延迟时间，而不是leader线程的其他take线程则需要等leader线程出队列了才唤醒其他take线程。

本文标题：阻塞队列—DelayedWorkQueue源码分析
本文网址：http://www.shufengxianlan.com/qtweb/news49/399299.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

响应式网站知识

行业网站建设

微信公众号开发　　 　绵阳网站建设　　 　成都网站建设公司　　 　成都酒店服定制　　 　成都托管服务器　　 　青羊区雕琢时光　　 　重庆网站建设　　 　简阳电信机房　　 　商城网站　　 　网站seo　　 　成都画册设计　　 　成都SEO优化公司　　 　东电技服　　 　东电技服　　 　托管服务器　　 　成都微信开发　　 　温江电信机房　　 　做移动网站　　 　成都做网站　　 　买链接