自学内容网 自学内容网
自学内容网 > 正文

PriorityQueue分析

🕗 发布于 2024-10-06 07:58 java  

概述

PriorityQueue，优先级队列，一种特殊的队列，作用是能保证每次取出的元素都是队列中权值最小的（Java的优先队列每次取最小元素，C++的优先队列每次取最大元素）。元素大小的评判可以通过元素本身的自然顺序(Natural Ordering)，也可通过构造时传入的比较器（如Java中的Comparator，C++的仿函数）。

Java中PriorityQueue通过二叉小顶堆实现，可用一棵完全二叉树表示。PriorityQueue实现Queue接口，不允许放入null元素；其通过堆实现，具体说是通过完全二叉树(Complete Binary Tree)实现的小顶堆（任意一个非叶子节点的权值，都不大于其左右子节点的权值），也就意味着可以通过数组来作为PriorityQueue的底层实现。
在这里插入图片描述

给每个元素按照层序遍历的方式进行编号，父子节点的编号之间的关系：

leftNo = parentNo*2+1
rightNo = parentNo*2+2
parentNo = (nodeNo-1)/2

通过上述三个公式，可以轻易计算出某个节点的父节点以及子节点的下标。这也就是为什么可以直接用数组来存储堆的原因。

应用场景

常用于需要按优先级处理元素的场景：

  • 任务调度：调度系统中的任务可能有不同的优先级。使用PriorityQueue可确保高优先级的任务先执行。JDK中的定时任务调度可能会使用类似的结构来管理不同的任务；
  • 图算法：许多图算法（如Dijkstra）需要频繁选择当前最小（或最大）权重的边或节点。PriorityQueue适合存储这些节点，以快速获取最优解；
  • 实时数据处理：在一些流式处理框架中，可能需要实时处理数据并保持数据的优先级。PriorityQueue可帮助维护这些数据的顺序；
  • MQ：在某些消息中间件中，可能会根据消息的优先级来处理消息。使用PriorityQueue可实现这一点，确保高优先级的消息先被消费；
  • 事件驱动架构：在事件驱动系统中，基于事件优先级，PriorityQueue可方便地管理和调度这些事件；
  • 合并多个有序序列：在处理多个已排序的数据流时，可使用PriorityQueue来合并这些序列，以保持整体的排序。

源码

PriorityQueue的peek()和element操作是常数时间，add()，offer()，无参数的remove()以及poll()方法的时间复杂度都是log(N)。

方法剖析
add()和offer()
都是向优先队列中插入元素，Queue接口规定二者对插入失败时的处理不同，前者在插入失败时抛出异常，后者则会返回false。

新加入的元素可能会破坏小顶堆的性质，因此需要进行必要的调整：

public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
modCount++;
int i = size;
if (i >= queue.length)
grow(i + 1);
siftUp(i, e);
size = i + 1;
return true;
}

扩容函数grow()类似于ArrayList里的grow()，再申请一个更大的数组，并将原数组的元素复制过去。siftUp(int k, E x)方法用于插入元素x并维持堆的特性。

private void siftUp(int k, E x) {
if (comparator != null)
siftUpUsingComparator(k, x, queue, comparator);
else
siftUpComparable(k, x, queue);
}

private static <T> void siftUpComparable(int k, T x, Object[] es) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = es[parent];
if (key.compareTo((T) e) >= 0)
break;
es[k] = e;
k = parent;
}
es[k] = key;
}

private static <T> void siftUpUsingComparator(
int k, T x, Object[] es, Comparator<? super T> cmp) {
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = es[parent];
if (cmp.compare(x, (T) e) >= 0)
break;
es[k] = e;
k = parent;
}
es[k] = x;
}

调整过程：从k指定的位置开始，将x逐层与当前点的parent进行比较并交换，直到满足x >= queue[parent]为止。比较可以是元素的自然顺序，也可以是依靠比较器的顺序。

element()和peek()
语义完全相同，都是获取但不删除队首元素，也就是队列中权值最小的那个元素，唯一的区别是当方法失败时前者抛出异常，后者返回null。根据小顶堆的性质，堆顶那个元素就是全局最小的那个；由于堆用数组表示，根据下标关系，0下标处的那个元素既是堆顶元素。所以直接返回数组0下标处的那个元素即可。
代码：

public E peek() {
return (E) queue[0];
}

remove()和poll()
语义完全相同，都是获取并删除队首元素，区别是当方法失败时前者抛出异常，后者返回null。由于删除操作会改变队列的结构，为维护小顶堆的性质，需要进行必要的调整。

public E poll() {
final Object[] es;
final E result;

if ((result = (E) ((es = queue)[0])) != null) {
modCount++;
final int n;
final E x = (E) es[(n = --size)];
es[n] = null;
if (n > 0) {
final Comparator<? super E> cmp;
if ((cmp = comparator) == null)
siftDownComparable(0, x, es, n);
else
siftDownUsingComparator(0, x, es, n, cmp);
}
}
return result;
}

首先记录0下标处的元素，并用最后一个元素替换0下标位置的元素，之后调用siftDown()方法对堆进行调整，最后返回原来0下标处的那个元素（也就是最小的那个元素）。siftDown(int k, E x)方法从k指定的位置开始，将x逐层向下与当前点的左右孩子中较小的那个交换，直到x小于或等于左右孩子中的任何一个为止。

private void siftDown(int k, E x) {
if (comparator != null)
siftDownUsingComparator(k, x, queue, size, comparator);
else
siftDownComparable(k, x, queue, size);
}

private static <T> void siftDownComparable(int k, T x, Object[] es, int n) {
// assert n > 0;
Comparable<? super T> key = (Comparable<? super T>)x;
int half = n >>> 1;           // loop while a non-leaf
while (k < half) {
int child = (k << 1) + 1; // assume left child is least
Object c = es[child];
int right = child + 1;
if (right < n && ((Comparable<? super T>) c).compareTo((T) es[right]) > 0)
c = es[child = right];
if (key.compareTo((T) c) <= 0)
break;
es[k] = c;
k = child;
}
es[k] = key;
}

private static <T> void siftDownUsingComparator(int k, T x, Object[] es, int n, Comparator<? super T> cmp) {
// assert n > 0;
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = es[child];
int right = child + 1;
if (right < n && cmp.compare((T) c, (T) es[right]) > 0)
c = es[child = right];
if (cmp.compare(x, (T) c) <= 0)
break;
es[k] = c;
k = child;
}
es[k] = x;
}

remove(Object o)
用于删除队列中跟o相等的某一个元素（如果有多个相等，只删除一个），源自Collection接口的方法。由于删除操作会改变队列结构，所以要进行调整；又由于删除元素的位置可能是任意的，所以调整过程比其它函数稍加繁琐。具体来说，remove(Object o)可分为2种情况：

  1. 删除的是最后一个元素。直接删除即可，不需调整；
  2. 不是最后一个元素，从删除点开始以最后一个元素为参照调用一次siftDown()即可。？？？
public boolean remove(Object o) {
// 通过遍历数组的方式找到第一个满足o.equals(queue[i])元素的下标
int i = indexOf(o);
if (i == -1)
return false;
else {
removeAt(i);
return true;
}
}

private int indexOf(Object o) {
if (o != null) {
final Object[] es = queue;
for (int i = 0, n = size; i < n; i++)
if (o.equals(es[i]))
return i;
}
return -1;
}

E removeAt(int i) {
// assert i >= 0 && i < size;
final Object[] es = queue;
modCount++;
int s = --size;
if (s == i) // removed last element
es[i] = null;
else {
E moved = (E) es[s];
es[s] = null;
siftDown(i, moved);
if (es[i] == moved) {
siftUp(i, moved);
if (es[i] != moved)
return moved;
}
}
return null;
}

堆排序

堆排序利用堆的特性来排序一个数组，通常包含以下几个步骤：

  • 构建最大堆
    将输入数据构建成一个最大堆，从最后一个非叶子节点开始，逐层向上调整堆，确保每个父节点都大于或等于子节点。
  • 排序过程
    • 将最大堆的根节点（最大值）与堆的最后一个元素交换，然后将堆的大小减一（实际上是忽略最后一个元素）；
    • 对新的根节点进行堆调整，重新建立最大堆；
    • 重复上述过程，直到堆的大小为1。

示例代码：

public class HeapSort {
public static void heapSort(int[] arr) {
PriorityQueue<Integer> maxHeap = new PriorityQueue<>((a, b) -> b - a);
for (int num : arr) {
    maxHeap.offer(num);
}
for (int i = 0; i < arr.length; i++) {
    arr[i] = maxHeap.poll();
}
}

public static void main(String[] args) {
int[] arr = {4, 10, 3, 5, 1};
heapSort(arr);
for (int num : arr) {
    System.out.print(num + " ");
}
}
}

拓展

JDK

JDK源码里基于PriorityQueue实现的数据结构主要有两个：

  • PriorityBlockingQueue：带优先级的阻塞队列
  • DelayQueue：延迟队列

PriorityBlockingQueue

在线程池里看到这个类，一个支持优先级排序的无界阻塞队列。PriorityBlockingQueue不会阻塞生产者，而只是在没有可消费的数据时阻塞消费者。因此使用时需要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，最终会耗尽所有可用的内存空间。

PriorityBlockingQueue的几个属性：

  • queue：数组，用来存放队列元素；
  • size：用来存放队列元素个数；
  • lock：独占锁对象，用来控制某个时间只能有一个线程可以进行入队、出队操作
  • notEmpty：条件(Condition)变量，用来实现take方法阻塞模式（跟其它阻塞队列相比，这里没有notFull条件变量，这是因为PriorityBlockingQueue是无界队列，其put方法是非阻塞的）。
  • allocationspinLock：是个自旋锁，使用CAS操作来保证某个时间只有一个线程可以扩容队列，状态为0或1，0表示当前没有进行扩容，1表示当前正在扩容；

PriorityBlockingQueue内部是使用平衡二叉树堆实现的，所以直接遍历队列元素不保证元素有序。

每次出队都返回优先级最高或最低的元素，默认使用对象的compareTo方法提供比较规则，这意味着队列元素必须实现Comparable接口；如果需要自定义比较规则则可以通过构造函数自定义comparator。

DelayQueue

ScheduledExecutorService

ScheduledExecutorService是JDK提供的用于定时任务调度API，支持设置任务优先级：

private ScheduledExecutorService schdExctr = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
    private ThreadFactory factory = Executors.defaultThreadFactory();

    @Override
    public Thread newThread(Runnable r) {
        Thread t = factory.newThread(r);
        t.setPriority(Thread.MIN_PRIORITY);
        return t;
    }
});

RocketMQ消息优先级

RocketMQ通过消息优先级实现消息的先入先出。消息优先级由生产者在发送消息时设置，范围为0-4，数字越大优先级越高。

Broker在接收到消息后，会将消息存储在不同的队列中，每种优先级对应一个队列。Broker会按照优先级从高到低的顺序消费队列中的消息，实现高优先级消息的先消费。

消费者在消费消息时，也会按照优先级从高到低的顺序拉取队列中的消息进行消费，保证高优先级消息的先消费。

RocketMQ通过设置消息优先级和隔离优先级消息到不同队列来实现消息的优先级，从而达到高优先级消息的先入先出。

仅供参考的代码：

// 生产者发送高优先级消息  
Message msg = new Message("TopicTest", "TagA", "KEY", "Hello".getBytes());
msg.setKeys("KEY1");
msg.setPriority(3);  // 设置消息优先级为3
producer.send(msg);

// 消费者设置消费消息的优先级  
pullConsumer.setConsumeMessageAck(true); 
pullConsumer.registerMessageListener(new PriorityMessageListener());

public class PriorityMessageListener implements MessageListener  {
@Override
public Action consumeMessage(MessageExt ext) { 
switch (ext.getPriority()) {
case 3: 
// 消费优先级为3的消息  
break;
}
} 
}

参考


原文地址：https://blog.csdn.net/lonelymanontheway/article/details/121961621

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

自学内容网
Copyright © 2015-2024