面試官：來(lái)了啊小伙子，以前經(jīng)常有小菜鳥被我虐個(gè)兩三輪就不敢來(lái)了胆胰，看你忍耐力還不錯(cuò)狞贱，以后應(yīng)該挺能加班的樣子。

Hydra：那可是蜀涨，我卷起來(lái)真的是連我自己都害怕跋规摇！

面試官：那咱們今天就繼續(xù)死磕隊(duì)列厚柳，聊聊PriorityBlockingQueue吧氧枣。

Hydra：沒(méi)問(wèn)題啊，PriorityBlockingQueue是一個(gè)支持優(yōu)先級(jí)的無(wú)界阻塞隊(duì)列别垮，之前介紹的隊(duì)列大多是FIFO先進(jìn)先出或LIFO后進(jìn)先出的便监，PriorityBlockingQueue不同，可以按照自然排序或自定義排序的順序在隊(duì)列中對(duì)元素進(jìn)行排序碳想。

我還是先寫一個(gè)例子吧烧董，使用offer方法向隊(duì)列中添加5個(gè)隨機(jī)數(shù)，然后使用poll方法從隊(duì)列中依次取出：

PriorityBlockingQueue<Integer> queue=new PriorityBlockingQueue<Integer>(5);

Random random = new Random();
System.out.println("add:");
for (int i = 0; i < 5; i++) {
    int j = random.nextInt(100);
    System.out.print(j+"  ");
    queue.offer(j);
}

System.out.println("\r\npoll:");
for (int i = 0; i < 5; i++) {
    System.out.print(queue.poll()+"  ");
}

查看運(yùn)行結(jié)果移袍，可以看到輸出順序與插入順序是不同的解藻，默認(rèn)情況下最終會(huì)按照自然排序的順序進(jìn)行輸出：

add:
68  34  40  31  44  
poll:
31  34  40  44  68 

PriorityBlockingQueue隊(duì)列就像下面這個(gè)神奇的容器，不管你按照什么順序往里塞數(shù)據(jù)葡盗，在取出的時(shí)候一定是按照排序完成后的順序出隊(duì)的螟左。

image

面試官：怎么感覺(jué)這功能有點(diǎn)雞肋啊啡浊，很多情況下我不想用自然排序怎么辦？

Hydra：一看你就沒(méi)仔細(xì)聽我前面講的胶背，除了自然排序外巷嚣，也可以自定義排序順序。如果我們想改變排序算法钳吟，也可以在構(gòu)造器中傳入一個(gè)Comparator對(duì)象廷粒，像下面這么一改就可以變成降序排序了：

PriorityBlockingQueue queue=new PriorityBlockingQueue<Integer>(10, new Comparator<Integer>() {
    @Override
    public int compare(Integer o1, Integer o2) {
        return o2-o1;
    }
});

面試官：我就隨口問(wèn)一句你還真以為我不知道啊，說(shuō)一下底層是怎么實(shí)現(xiàn)的吧红且？

Hydra：在講底層的原理之前坝茎，就不得不先提一下二叉堆的數(shù)據(jù)結(jié)構(gòu)了。二叉堆是一種特殊的堆暇番，它的結(jié)構(gòu)和完全二叉樹非常類似嗤放。如果父節(jié)點(diǎn)的值總小于子節(jié)點(diǎn)的值，那么它就是一個(gè)最小二叉堆壁酬，反之則是最大二叉堆次酌，并且每個(gè)節(jié)點(diǎn)的左子樹和右子樹也是一個(gè)二叉堆。

以一個(gè)最小二叉堆為例：

image

這個(gè)最小二叉堆保存在數(shù)組中的順序是這樣的：

[1,2,3,4,5,6,7,8,9]

根據(jù)它的特性舆乔，可以輕松的計(jì)算出一個(gè)節(jié)點(diǎn)的父節(jié)點(diǎn)或子節(jié)點(diǎn)在數(shù)組中對(duì)應(yīng)的位置岳服。假設(shè)一個(gè)元素在數(shù)組中的下標(biāo)是t，那么父節(jié)點(diǎn)希俩、左右子節(jié)點(diǎn)的下標(biāo)計(jì)算公式如下：

parent(t) = (t - 1) >>> 1 
left(t) = t << 1 + 1
right(t) = t << 1 + 2

以上面的二叉堆中的元素6為例吊宋，它在數(shù)組中的下標(biāo)是5，可以計(jì)算出它的父節(jié)點(diǎn)下標(biāo)為2斜纪，對(duì)應(yīng)元素為3：

parent(5) = 100 >>> 1 = 2

如果要計(jì)算元素4的左右子節(jié)點(diǎn)的話贫母，它的下標(biāo)是3，計(jì)算出的子節(jié)點(diǎn)坐標(biāo)分別為7,8盒刚，對(duì)應(yīng)的元素為8,9：

left(3) = 11 << 1 + 1 = 7
right(3) = 11 << 1 + 2 = 8

在上面計(jì)算元素的數(shù)組位置過(guò)程中使用了左移右移操作腺劣，是不是感覺(jué)非常酷炫因块？

面試官：行了別貧了橘原，鋪墊了半點(diǎn)，趕緊說(shuō)隊(duì)列的底層原理涡上。

Hydra：別急趾断，下面就講了，在PriorityBlockingQueue中吩愧，關(guān)鍵的屬性有下面這些：

private transient Object[] queue;
private transient int size;
private transient Comparator<? super E> comparator;
private final ReentrantLock lock;
private final Condition notEmpty;

前面我們也說(shuō)了芋酌，二叉堆可以用數(shù)組的形式存儲(chǔ)，所以隊(duì)列的底層仍然是使用數(shù)組來(lái)存放元素的雁佳。在無(wú)參構(gòu)造函數(shù)中脐帝，隊(duì)列的初始容量是11同云，comparator為空，也就是使用元素自身的compareTo方法來(lái)進(jìn)行比較排序堵腹。和ArrayBlockingQueue類似炸站，底層通過(guò)ReentrantLock實(shí)現(xiàn)線程間的并發(fā)控制, 并使用Condition實(shí)現(xiàn)線程的等待及喚醒。

面試官：這么一看疚顷，屬性和ArrayBlockingQueue還真是基本差不多啊旱易，那結(jié)構(gòu)就介紹到這吧，說(shuō)重點(diǎn)腿堤，元素是怎么按照排序方法插入的阀坏？

Hydra：我們先對(duì)offer方法的執(zhí)行流程進(jìn)行分析，如果隊(duì)列中元素未滿笆檀，且在默認(rèn)情況下comparator為空時(shí)全释，按照自然順序排序，會(huì)執(zhí)行siftUpComparable方法：

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

如果隊(duì)列為空误债，那么元素直接入隊(duì)，如果隊(duì)列中已經(jīng)有元素了妄迁，那么就需要判斷插入的位置了寝蹈。首先獲取父節(jié)點(diǎn)的坐標(biāo)，將自己的值和父節(jié)點(diǎn)進(jìn)行比較登淘，可以分為兩種情況：

• 如果新節(jié)點(diǎn)的值比父節(jié)點(diǎn)大箫老，那么說(shuō)明當(dāng)前父節(jié)點(diǎn)就是較小的元素，不需要進(jìn)行調(diào)整黔州，直接將元素添加到隊(duì)尾
• 如果新節(jié)點(diǎn)的值比父節(jié)點(diǎn)小的話耍鬓，那么就要進(jìn)行上浮操作。先將父節(jié)點(diǎn)的值復(fù)制到子節(jié)點(diǎn)的位置流妻，下一次將新節(jié)點(diǎn)的值與父節(jié)點(diǎn)的父節(jié)點(diǎn)進(jìn)行比較牲蜀。這一上浮過(guò)程會(huì)持續(xù)進(jìn)行，直到新節(jié)點(diǎn)的值比父節(jié)點(diǎn)大绅这，或新節(jié)點(diǎn)上浮成為根節(jié)點(diǎn)為止

還是以上面數(shù)據(jù)插入過(guò)程為例涣达，來(lái)演示二叉樹的構(gòu)建過(guò)程：

image

在將新元素添加到隊(duì)列中后，隊(duì)列中元素的計(jì)數(shù)加1证薇，并且去喚醒阻塞在notEmpty上的等待線程度苔。

面試官：那么如果不是自然排序的時(shí)候，邏輯會(huì)發(fā)生改變嗎浑度？

Hydra：如果comparator不為空的話寇窑，邏輯與上面的方法基本一致，唯一不同的是在進(jìn)行比較時(shí)調(diào)用的是傳入的自定義comparator的compare方法箩张。

面試官：剛才你在講offer方法的時(shí)候甩骏，強(qiáng)調(diào)了隊(duì)列中元素未滿這一個(gè)條件窗市，開始的時(shí)候不是說(shuō)PriorityBlockingQueue是一個(gè)無(wú)界隊(duì)列么，那為什么還要加這一個(gè)條件横漏？

Hydra：雖然說(shuō)它是一個(gè)無(wú)界隊(duì)列谨设，但其實(shí)隊(duì)列的長(zhǎng)度上限是Integer.MAX_VALUE - 8，并且底層是使用的數(shù)組保存元素缎浇，在初始化數(shù)組的時(shí)候也會(huì)指定一個(gè)長(zhǎng)度扎拣，如果超過(guò)這個(gè)長(zhǎng)度的話，那么就需要進(jìn)行擴(kuò)容素跺，執(zhí)行tryGrow方法：

private void tryGrow(Object[] array, int oldCap) {
    lock.unlock(); // 釋放鎖
    Object[] newArray = null;
    if (allocationSpinLock == 0 &&
        //cas 加鎖
        UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,0, 1)) {
        try {
            //計(jì)算擴(kuò)容后的容量
            int newCap = oldCap + ((oldCap < 64) ?
                                   (oldCap + 2) : // grow faster if small
                                   (oldCap >> 1));
            // 避免超出上限
            if (newCap - MAX_ARRAY_SIZE > 0) {    
                int minCap = oldCap + 1;
                if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
                    throw new OutOfMemoryError();
                newCap = MAX_ARRAY_SIZE;
            }
            if (newCap > oldCap && queue == array)
                //申請(qǐng)新的數(shù)組
                newArray = new Object[newCap];
        } finally {
            //釋放cas鎖標(biāo)志位
            allocationSpinLock = 0;
        }
    }
    //其他線程正在擴(kuò)容二蓝，讓出CPU
    if (newArray == null) // back off if another thread is allocating
        Thread.yield();
    //加獨(dú)占式鎖，拷貝原先隊(duì)列中的數(shù)據(jù)
    lock.lock();
    if (newArray != null && queue == array) {
        queue = newArray;
        System.arraycopy(array, 0, newArray, 0, oldCap);
    }
}

先說(shuō)鎖的操作指厌，在進(jìn)行擴(kuò)容前刊愚，會(huì)先釋放獨(dú)占式的lock，因?yàn)閿U(kuò)容操作需要一定的時(shí)間踩验，如果在這段時(shí)間內(nèi)還持有鎖的話會(huì)降低隊(duì)列的吞吐量鸥诽。因此這里使用cas的方式保證擴(kuò)容這一操作本身是排他性的，即只有一個(gè)線程來(lái)實(shí)現(xiàn)擴(kuò)容箕憾。在完成新數(shù)組的申請(qǐng)后牡借，會(huì)釋放cas鎖的標(biāo)志位，并在拷貝隊(duì)列中原有數(shù)據(jù)到新數(shù)組前袭异，再次加獨(dú)占式鎖lock钠龙，保證線程間的數(shù)據(jù)安全。

至于擴(kuò)容操作也很簡(jiǎn)單御铃，假設(shè)當(dāng)前數(shù)組長(zhǎng)度為n碴里，如果小于64的話那么數(shù)組長(zhǎng)度擴(kuò)為2n+2，如果大于64則擴(kuò)為1.5n上真，并且擴(kuò)容后的數(shù)組不能超過(guò)上面說(shuō)的上限值咬腋。申請(qǐng)完成新的數(shù)組空間后，使用native方法實(shí)現(xiàn)數(shù)據(jù)的拷貝谷羞。

假設(shè)初始長(zhǎng)度為5帝火，當(dāng)有新元素要入隊(duì)時(shí)，就需要進(jìn)行擴(kuò)容湃缎，如圖所示：

image

面試官：ok犀填，講的還不賴，該說(shuō)出隊(duì)的方法了吧嗓违？

Hydra：嗯九巡，有了前面的基礎(chǔ)，出隊(duì)過(guò)程理解起來(lái)也非常簡(jiǎn)單蹂季，還是以自然排序?yàn)槔峁悖匆幌?code>dequeue方法（省略了部分不重要的代碼）：

private E dequeue() {
    int n = size - 1;
    // ...
    Object[] array = queue;
    E result = (E) array[0];
    E x = (E) array[n];
    array[n] = null;
    // ...
    siftDownComparable(0, x, array, n);
    // ...
    size = n;
    return result;    
}

如果隊(duì)列為空疏日，dequeue方法會(huì)直接返回null，否則返回?cái)?shù)組中的第一個(gè)元素撒汉。在將隊(duì)尾元素保存后沟优，清除隊(duì)尾節(jié)點(diǎn)，然后調(diào)用siftDownComparable方法睬辐，調(diào)整二叉堆的結(jié)構(gòu)挠阁，使其成為一個(gè)新的最小二叉堆：

private static <T> void siftDownComparable(int k, T x, Object[] array,int n) {
    if (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 = array[child];
            int right = child + 1;
            if (right < n &&
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                c = array[child = right];
            if (key.compareTo((T) c) <= 0)
                break;
            array[k] = c;
            k = child;
        }
        array[k] = key;
    }
}

首先解釋一下half的作用，它用來(lái)尋找隊(duì)列的中間節(jié)點(diǎn)溯饵，所有非葉子節(jié)點(diǎn)的坐標(biāo)都不會(huì)超過(guò)這個(gè)half值侵俗。分別以樹中含有奇數(shù)個(gè)節(jié)點(diǎn)和偶數(shù)個(gè)節(jié)點(diǎn)為例：

image

[n=9]  1001 >>> 1 =100 =4
[n=8]  1000 >>> 1 =100 =4

可以看到，奇數(shù)和偶數(shù)的情況下計(jì)算出的half值都是4丰刊，即非葉子節(jié)點(diǎn)的下標(biāo)不會(huì)超過(guò)4隘谣，對(duì)應(yīng)上圖中的元素為5叨叙。

面試官：計(jì)算二叉樹最后非葉子節(jié)點(diǎn)坐標(biāo)這點(diǎn)知識(shí)匈挖，大一學(xué)過(guò)數(shù)據(jù)結(jié)構(gòu)的新生都知道，趕緊說(shuō)正題祷舀！

Hydra：著什么急啊秩仆，前面我們也說(shuō)了熄求，在將堆頂元素取出后，堆頂位置的元素出現(xiàn)空缺逗概，需要調(diào)整堆結(jié)構(gòu)使二叉堆的結(jié)構(gòu)特性保持不變。這時(shí)候比較簡(jiǎn)單的方法就是將尾結(jié)點(diǎn)直接填充到堆頂忘衍，然后從堆頂開始調(diào)整結(jié)構(gòu)逾苫。

因此在代碼中，每次執(zhí)行堆頂節(jié)點(diǎn)的出隊(duì)后枚钓，都將尾節(jié)點(diǎn)取出铅搓，然后從根節(jié)點(diǎn)開始向下比較，這一過(guò)程可以稱為下沉搀捷。下沉過(guò)程從根節(jié)點(diǎn)開始星掰，首先獲取左右子節(jié)點(diǎn)的坐標(biāo)，并取出存儲(chǔ)的元素值較小的那個(gè)嫩舟，和key進(jìn)行比較：

• 如果key比左右節(jié)點(diǎn)都要小氢烘，那么說(shuō)明找到了位置，比較結(jié)束家厌，直接使用它替換父節(jié)點(diǎn)即可
• 否則的話播玖，調(diào)整二叉堆結(jié)構(gòu)，將較小的子節(jié)點(diǎn)上浮饭于，使用它替換父節(jié)點(diǎn)蜀踏。然后將用于比較的父節(jié)點(diǎn)坐標(biāo)k下移調(diào)整為較小子節(jié)點(diǎn)维蒙，準(zhǔn)備進(jìn)行下一次的比較

別看我白話這么一大段，估計(jì)你還是不明白果覆，給你畫個(gè)圖吧颅痊，以上面的隊(duì)列執(zhí)行一次poll方法為例：

image

后面的操作也是以此類推，分析到這出隊(duì)操作也就結(jié)束了局待，PriorityBlockingQueue也沒(méi)什么其他好講的了斑响。

面試官：我發(fā)現(xiàn)你現(xiàn)在開始偷懶了，前面的面試?yán)锬氵€分一下阻塞和非阻塞方法燎猛，現(xiàn)在不說(shuō)一下這兩種方式的區(qū)別就想蒙混過(guò)關(guān)了恋捆？

Hydra：嗨，在PriorityBlockingQueue里阻塞和非阻塞的區(qū)別其實(shí)并不大重绷，首先因?yàn)樗且粋€(gè)無(wú)界的隊(duì)列沸停，因此添加元素的操作是不會(huì)被阻塞的，如果看一下源碼昭卓，你就會(huì)發(fā)現(xiàn)其他的添加方法add愤钾、put也是直接調(diào)用的offer方法。

而取出元素操作會(huì)受限制于隊(duì)列是否為空候醒，因此可能會(huì)發(fā)生阻塞能颁，阻塞方法take和非阻塞的poll會(huì)稍有不同，如果出現(xiàn)隊(duì)列為空的情況倒淫，poll會(huì)直接返回null伙菊，而take會(huì)將線程在notEmpty上進(jìn)行阻塞，等待隊(duì)列中被添加元素后喚醒敌土。

面試官：嗯镜硕，優(yōu)先級(jí)隊(duì)列我們也聊的差不多了，反正都聊了這么久的隊(duì)列了返干，不介意我們把剩余的幾個(gè)也說(shuō)完吧兴枯？

Hydra：沒(méi)問(wèn)題啊，畢竟我能有什么選擇呢矩欠？

image

最后

如果覺(jué)得對(duì)您有所幫助财剖，小伙伴們可以點(diǎn)贊、轉(zhuǎn)發(fā)一下~非常感謝

微信搜索：碼農(nóng)參上癌淮，來(lái)加個(gè)好友躺坟，點(diǎn)贊之交也好啊~

公眾號(hào)后臺(tái)回復(fù)“面試”、“導(dǎo)圖”乳蓄、“架構(gòu)”瞳氓、“實(shí)戰(zhàn)”，獲得免費(fèi)資料哦~