目录

  • 什么是堆?

    • 满二叉树
    • 完全二叉树
  • 堆的实现
    • 最大堆的插入(ADD)
    • 最大堆的删除(DELETE)
  • DelayedWorkQueue类
    • 属性
    • 插入元素方法
    • 等待获取队列头元素
    • 超时等待获取队列头元素
  • 推荐博客
  • 总结

正文

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

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

实现优先级队列高效常用的一种方式就是使用堆。

回到顶部

什么是堆?

堆通常是一个可以被看做一棵树的数组对象。

堆(heap)又被为优先队列(priority queue)。尽管名为优先队列,但堆并不是队列。

因为队列中允许的操作是先进先出(FIFO),在队尾插入元素,在队头取出元素。

而堆虽然在堆底插入元素,在堆顶取出元素,但是堆中元素的排列不是按照到来的先后顺序,而是按照一定的优先顺序排列的。

这里来说明一下满二叉树的概念与完全二叉树的概念。

满二叉树

  除了叶子节点,所有的节点的左右孩子都不为空,就是一棵满二叉树,如下图。

可以看出:满二叉树所有的节点都拥有左孩子,又拥有右孩子。

完全二叉树

  不一定是一个满二叉树,但它不满的那部分一定在右下侧,如下图

堆总是满足下列性质:

  • 堆中某个节点的值总是不大于或不小于其父节点的值;

  • 堆总是一棵完全二叉树。

  • 最大值时,称为“最大堆”,也称大顶堆;
  • 最小值时,称为“最小堆”,也称小顶堆。

回到顶部

堆的实现

堆是一个二叉树,但是它最简单的方式是通过数组去实现二叉树,而且因为堆是一个完全二叉树,就不存在数组空间的浪费。怎么使用数组来存储二叉树呢?

就是用数组的下标来模拟二叉树的各个节点,比如说根节点就是0,第一层的左节点是1,右节点是2。由此我们可以得出下列公式:

1 // 对于n位置的节点来说:
2 int left = 2 * n + 1; // 左子节点
3 int right = 2 * n + 2; // 右子节点
4 int parent = (n - 1) / 2; // 父节点,当然n要大于0,根节点是没有父节点的

对于堆来说,只有两个操作,插入insert和删除remove,不管插入还是删除保证堆的成立条件,1.是完全二叉树,2.父节点的值不能小于子节点的值。

最大堆的插入(ADD)

1 public void insert(int value) {2      // 第一步将插入的值,直接放在最后一个位置。并将长度加一3      store[size++] = value;4      // 得到新插入值所在位置。5      int index = size - 1;6      while(index > 0) {7          // 它的父节点位置坐标8          int parentIndex = (index - 1) / 2;9          // 如果父节点的值小于子节点的值,你不满足堆的条件,那么就交换值
10          if (store[index] > store[parentIndex]) {
11              swap(store, index, parentIndex);
12              index = parentIndex;
13          } else {
14              // 否则表示这条路径上的值已经满足降序,跳出循环
15              break;
16          }
17      }
18 }

主要步骤:

  • 直接将value插入到size位置,并将size自增,这样store数组中插入一个值了。

  • 要保证从这个叶节点到根节点这条路径上的节点,满足父节点的值不能小于子节点。

  • 通过int parentIndex = (index - 1) / 2得到父节点,如果比父节点值大,那么两者位置的值交换,然后再拿这个父节点和它的父父节点比较。

    直到这个节点值比父节点值小,或者这个节点已经是根节点就退出循环。

因为每次循环index都是除以2这种倍数递减的方式,所以它最多循环次数是(log N)次。

最大堆的删除(DELETE)

1 public int remove() {2       // 将根的值记录,最后返回3       int result = store[0];4       // 将最后位置的值放到根节点位置5       store[0] = store[--size];6       int index = 0;7       // 通过循环,保证父节点的值不能小于子节点。8       while(true) {9           int leftIndex = 2 * index + 1; // 左子节点
10           int rightIndex = 2 * index + 2; // 右子节点
11           // leftIndex >= size 表示这个子节点还没有值。
12           if (leftIndex >= size) break;
13           int maxIndex = leftIndex;
14           //找到左右节点中较大的一个节点
15           if (store[leftIndex] < store[rightIndex]) maxIndex = rightIndex;
16           //与子节点中较大的子节点比较,如果子节点更大,则交换位置
17           //为什么要与较大的子节点比较呢?如果和较小的节点比较,没有交换位置,但有可能比较大的节点小
18           if (store[index] < store[maxIndex]) {
19               swap(store, index, maxIndex);
20               index = maxIndex;
21           } else {
22               //满足子节点比当前节点小,退出循环
23               break;
24           }
25       }
26       //返回最开始的第一个值
27       return result;
28 }

在堆中最大值就在根节点,所以操作步骤:

  1. 将根节点的值保存到result中。

  2. 将最后节点的值移动到根节点,再将长度减一,这样满足堆成立第一个条件,堆是一个完全二叉树。

  3. 使用循环,来满足堆成立的第二个条件,父节点的值不能小于子节点的值。

  4. 最后返回result。

每次循环我们都是以2的倍数递增,所以它也是最多循环次数是(log N)次。

所以通过堆这种方式可以快速实现优先级队列,它的插入和删除操作的效率都是O(log N)。

那么怎么实现堆排序?这个很简单,利用优先队列的特性:

  1. 先遍历数组。将数组中的值依次插入到堆中。
  2. 然后再用一个循环将值从堆中取出来。

1 private static void headSort(int[] arr) {2       int size = arr.length;3       Head head = new Head(size);4       for (int i = 0; i < size; i++) {5           head.insert(arr[i]);6       }7       for (int i = 0; i < size; i++) {8           //  实现从大到小的排序9           arr[size - 1 - i] = head.remove();
10       }
11 }

堆排序的效率:因为每次插入数据效率是O(log N),而我们需要进行n次循环,将数组中每个值插入到堆中,所以它的执行时间是O(N * log N)级。

回到顶部

DelayedWorkQueue类

1 static class DelayedWorkQueue extends AbstractQueue<Runnable>
2         implements BlockingQueue<Runnable> {

从定义中看出DelayedWorkQueue是一个阻塞队列。并且DelayedWorkQueue是一个最小堆,最顶点的值最小,即堆中某个节点的值总是不小于其父节点的值。

属性

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

DelayedWorkQueue是用数组来储存队列中的元素,那么我们看看它是怎么实现优先级队列的。

插入元素方法

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

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

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

数组扩容方法:

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

插入元素排序siftUp方法:

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

主要是三步:

  • 元素个数超过数组长度,就会调用grow()方法,进行数组扩容。
  • 将新元素e添加到优先级队列中对应的位置,通过siftUp方法,保证按照元素的优先级排序。
  • 如果新插入的元素是队列头,即更换了队列头,那么就要唤醒正在等待获取任务的线程。这些线程可能是因为原队列头元素的延时时间没到,而等待的。

我们来看看动画

假设现有元素 5 需要插入,为了维持完全二叉树的特性,新插入的元素一定是放在结点 6 的右子树;同时为了满足任一结点的值要小于左右子树的值这一特性,新插入的元素要和其父结点作比较,如果比父结点小,就要把父结点拉下来顶替当前结点的位置,自己则依次不断向上寻找,找到比自己大的父结点就拉下来,直到没有符合条件的值为止。

动画讲解:

  1. 在这里先将元素 5 插入到末尾,即放在结点 6 的右子树。

  2. 然后与父类比较, 6 > 5 ,父类数字大于子类数字,子类与父类交换。

  3. 重复此操作,直到不发生替换。

立即获取队列头元素

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

1 public long getDelay(TimeUnit unit) {
2     return unit.convert(time - now(), NANOSECONDS);
3 }

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

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

这个方法与我们在第一节中,介绍堆的删除方法一样。

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

移除元素排序siftDown方法:

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

我们来看看动画

核心点:将最后一个元素填充到堆顶,然后不断的下沉这个元素。

假设要从节点 1 ,也可以称为取出节点 1 ,为了维持完全二叉树的特性 ,我们将最后一个元素 6 去替代这个 1 ;然后比较 1 和其子树的大小关系,如果比左右子树大(如果存在的话),就要从左右子树中找一个较小的值替换它,而它能自己就要跑到对应子树的位置,再次循环这种操作,直到没有子树比它小。

通过这样的操作,堆依然是堆,总结一下:

  • 找到要删除的节点(取出的节点)在数组中的位置
  • 用数组中最后一个元素替代这个位置的元素
  • 当前位置和其左右子树比较,保证符合最小堆的节点间规则
  • 删除最后一个元素

等待获取队列头元素

1 public RunnableScheduledFuture<?> take() throws InterruptedException {2     final ReentrantLock lock = this.lock;3     lock.lockInterruptibly();4     try {5         for (;;) {6             RunnableScheduledFuture<?> first = queue[0];7             // 如果没有任务,就让线程在available条件下等待。8             if (first == null)9                 available.await();
10             else {
11                 // 获取任务的剩余延时时间
12                 long delay = first.getDelay(NANOSECONDS);
13                 // 如果延时时间到了,就返回这个任务,用来执行。
14                 if (delay <= 0)
15                     return finishPoll(first);
16                 // 将first设置为null,当线程等待时,不持有first的引用
17                 first = null; // don't retain ref while waiting
18
19                 // 如果还是原来那个等待队列头任务的线程,
20                 // 说明队列头任务的延时时间还没有到,继续等待。
21                 if (leader != null)
22                     available.await();
23                 else {
24                     // 记录一下当前等待队列头任务的线程
25                     Thread thisThread = Thread.currentThread();
26                     leader = thisThread;
27                     try {
28                         // 当任务的延时时间到了时,能够自动超时唤醒。
29                         available.awaitNanos(delay);
30                     } finally {
31                         if (leader == thisThread)
32                             leader = null;
33                     }
34                 }
35             }
36         }
37     } finally {
38         if (leader == null && queue[0] != null)
39             // 唤醒等待任务的线程
40             available.signal();
41         lock.unlock();
42     }
43 }

如果队列中没有任务,那么就让当前线程在available条件下等待。如果队列头任务的剩余延时时间delay大于0,那么就让当前线程在available条件下等待delay时间。

超时等待获取队列头元素

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

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

回到顶部

推荐博客

  https://www.cnblogs.com/chen-haozi/p/10227797.html

回到顶部

总结

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

转载于:https://my.oschina.net/tingzi/blog/3028651

并发编程(十四)—— ScheduledThreadPoolExecutor 实现原理与源码深度解析 之DelayWorkQueue...相关推荐

  1. java并发编程——线程池的工作原理与源码解读

    2019独角兽企业重金招聘Python工程师标准>>> 线程池的简单介绍 基于多核CPU的发展,使得多线程开发日趋流行.然而线程的创建和销毁,都涉及到系统调用,比较消耗系统资源,所以 ...

  2. JVM CPU Profiler技术原理及源码深度解析

    本文介绍了JVM平台上CPU Profiler的实现原理,希望能帮助读者在使用类似工具的同时也能清楚其内部的技术实现. 引言 研发人员在遇到线上报警或需要优化系统性能时,常常需要分析程序运行行为和性能 ...

  3. 并发编程之 Executor 线程池原理与源码解读

    并发编程之 Executor 线程池原理与源码解读 线程是调度 CPU 资源的最小单位,线程模型分为 KLT 模型与 ULT 模型,JVM使用的是 KLT 模型.java线程与 OS 线程保持 1:1 ...

  4. 《Spring源码深度解析 郝佳 第2版》SpringBoot体系分析、Starter的原理

    往期博客 <Spring源码深度解析 郝佳 第2版>容器的基本实现与XML文件的加载 <Spring源码深度解析 郝佳 第2版>XML标签的解析 <Spring源码深度解 ...

  5. 《Spring源码深度解析 郝佳 第2版》JDBC、MyBatis原理

    往期博客 <Spring源码深度解析 郝佳 第2版>容器的基本实现与XML文件的加载 <Spring源码深度解析 郝佳 第2版>XML标签的解析 <Spring源码深度解 ...

  6. RocketMQ源码(十)—Broker 消息刷盘服务GroupCommitService、FlushRealTimeService、CommitRealTimeService源码深度解析

    深入的介绍了broker的消息刷盘服务源码解析,以及高性能的刷盘机制. 学习RocketMQ的时候,我们知道RocketMQ的刷盘策略有两个,同步或者是异步: 1. 同步刷盘:如上图所示,只有消息真正 ...

  7. hibernate 并发获取session失败 空指针_高并发之|通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程...

    核心逻辑概述 ThreadPoolExecutor是Java线程池中最核心的类之一,它能够保证线程池按照正常的业务逻辑执行任务,并通过原子方式更新线程池每个阶段的状态. ThreadPoolExecu ...

  8. 并发编程之Executor线程池原理与源码解读

    1. 线程池 "线程池",顾名思义就是一个线程缓存,线程是稀缺资源,如果被无限制的创建,不 仅会消耗系统资源,还会降低系统的稳定性,因此Java中提供线程池对线程进行统一分配. 调 ...

  9. 《视觉SLAM十四讲》第二版源码slambook2编译调试

    slambook2-master/ch2 编译正常,log如下: slambook2-master/ch2# mkdir build && cd build && cm ...

  10. Spring源码深度解析(郝佳)-学习-Spring Boot体系原理

      Spring Boot是由Pivotal团队提供的全新框架,其设计目的用来简化新Spring应用初始化搭建以及开发过程,该框架使用了我写的方式进行配置,从而开发人员不再需要定义样板化的配置,通过这 ...

最新文章

  1. POJ 3347 Kadj Squares(复杂的线段相交问题)
  2. wxWidgets:弹出 wxWidgets 示例
  3. 动态生成数据后绑定事件
  4. Third week-homework(员工管理系统)
  5. 项目的启动顺序_多个项目进行如何做好进度管理
  6. 高颜值可定制在线作图工具-第二版
  7. python多线程有用吗_Python多线程理解
  8. 浏览器怎么清理缓存_Mac系统浏览器缓存清理工具
  9. 【C++基础学习】二维数组的动态分配及参数传递
  10. Service获取客户端IP地址(java)
  11. 操作系统—进程的定义、组成和组织方式(思维导图)
  12. WEB标准 基础(一) 到底是什么?
  13. CentOS7 配置免密登陆
  14. CTF-misc练习(https://buuoj.cn)之第二页
  15. 共享充电宝为啥能够盈利
  16. 学习游戏开发,有哪些常识需要了解
  17. 【VisionMaster】二次开发之第三方库的使用
  18. 设计模式01策略模式
  19. GCC 编译器警告——【-Wunused-variable】【-Wunused-parameter】
  20. 500是什么php,HTTP 500,该怎么解决

热门文章

  1. 03day输入及转义符的使用
  2. 仓库建设-斜率优化DP
  3. [tldk][dpdk][dev] TLDK--基于dpdk的用户态协议栈传输层组件简单调研
  4. HTML5微数据学习笔记
  5. 根据自身工作经验总结的一个工作问题解决思路
  6. Asp.net页面之间传“大量的参数”
  7. 内容超过7行显示查看全文
  8. oracle弱口令攻击
  9. mybatis结果的组装(springboot)
  10. sql server安装