PriorityBlockingQueue<E> 是 Java 并發(fā)包（java.util.concurrent）中提供的一個(gè)線程安全的、無界、優(yōu)先級(jí)隊(duì)列。它的核心思想是：

每次取出的元素，都是當(dāng)前隊(duì)列中“優(yōu)先級(jí)最高”的那個(gè)元素（即最小值，依據(jù)自然排序或自定義比較器）。

一、關(guān)鍵特性總結(jié)

二、核心機(jī)制解析

1.底層數(shù)據(jù)結(jié)構(gòu)：二叉最小堆

• 使用 Object[] queue 存儲(chǔ)元素。
• 對(duì)于任意節(jié)點(diǎn) i：
  • 左孩子：2*i + 1
  • 右孩子：2*i + 2
  • 父節(jié)點(diǎn)：(i - 1) / 2
• 堆性質(zhì)：父節(jié)點(diǎn) ≤ 子節(jié)點(diǎn) → 根節(jié)點(diǎn)（queue[0]）是最小值（最高優(yōu)先級(jí)）。

2.擴(kuò)容機(jī)制（tryGrow）

• 當(dāng)數(shù)組滿時(shí)，自動(dòng)擴(kuò)容：
  • 小容量（<64）：增長較快（+ oldCap + 2）
  • 大容量：增長 50%（oldCap >> 1）
• 特殊設(shè)計(jì)：擴(kuò)容時(shí)不持有主鎖（lock），而是用 CAS 自旋鎖（allocationSpinLock）避免阻塞消費(fèi)者。
  • 目的：防止生產(chǎn)者擴(kuò)容時(shí)長時(shí)間持有鎖，導(dǎo)致消費(fèi)者“餓死”。

3.堆調(diào)整操作

• siftUp：插入新元素后，從底部向上調(diào)整（冒泡到合適位置）。
• siftDown：刪除根節(jié)點(diǎn)后，把最后一個(gè)元素放到根，再向下調(diào)整。
• 分為兩種版本：
  • siftUpComparable / siftDownComparable：使用元素自身的 compareTo()
  • siftUpUsingComparator / siftDownUsingComparator：使用外部 Comparator

4.構(gòu)造函數(shù)邏輯

• 如果傳入的是 SortedSet 或 PriorityQueue，直接復(fù)用其排序規(guī)則（無需重新建堆）。
• 否則，對(duì)傳入集合調(diào)用 heapify() 從底向上建堆（時(shí)間復(fù)雜度 O(n)）。

5.阻塞與非阻塞操作

6.關(guān)于迭代器（重要！）

Iterator<E> it = pq.iterator();
// ? 不保證按優(yōu)先級(jí)順序遍歷！

• 原因：堆的數(shù)組存儲(chǔ)不是排序數(shù)組，只是滿足堆性質(zhì)。
• 正確做法：如需有序遍歷，必須：

  Object[] arr = pq.toArray();
  Arrays.sort(arr); // 或使用 Comparator
  

三、FIFOEntry 示例：解決“優(yōu)先級(jí)相同時(shí)的公平性”

當(dāng)多個(gè)元素優(yōu)先級(jí)相同（compareTo == 0），默認(rèn)不保證誰先出隊(duì)。

解決方案：引入“插入順序”作為第二排序鍵。

class FIFOEntry<E extends Comparable<? super E>>
    implements Comparable<FIFOEntry<E>> {
  static final AtomicLong seq = new AtomicLong(0);
  final long seqNum;   // 插入序號(hào)
  final E entry;
  public int compareTo(FIFOEntry<E> other) {
    int res = entry.compareTo(other.entry);
    if (res == 0)
      res = Long.compare(seqNum, other.seqNum); // 先插入的先出
    return res;
  }
}

使用時(shí)：pq.offer(new FIFOEntry(myElement));

四、典型使用場景

• 任務(wù)調(diào)度系統(tǒng)：高優(yōu)先級(jí)任務(wù)先執(zhí)行。
• 事件處理：緊急事件優(yōu)先處理。
• 合并多個(gè)有序流：如多路歸并（配合 take() 阻塞特性）。

五、注意事項(xiàng)

1. 不要依賴 iterator() 的順序！
2. 避免在比較器中拋異常：會(huì)導(dǎo)致隊(duì)列狀態(tài)不一致。
3. 內(nèi)存風(fēng)險(xiǎn)：雖然是“無界”，但大量積壓會(huì)導(dǎo)致 OOM。
4. 性能：高并發(fā)下，所有操作串行化（單鎖），吞吐量不如 ConcurrentLinkedQueue，但語義不同。

總結(jié)

PriorityBlockingQueue = 線程安全的 PriorityQueue + BlockingQueue 接口
它適合需要按優(yōu)先級(jí)消費(fèi)、且允許多線程協(xié)作的場景，但要注意其無界性和迭代無序性。

如果你理解了二叉堆、CAS 自旋鎖、以及阻塞條件（Condition notEmpty），就掌握了它的精髓。

到此這篇關(guān)于Java并發(fā)包中的PriorityBlockingQueue解析的文章就介紹到這了,更多相關(guān)java并發(fā)包PriorityBlockingQueue內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論