[[401100]]

面試官：好了，聊完了ArrayBlockingQueue，我們接著說(shuō)說(shuō)LinkedBlockingQueue吧

Hydra：還真是不給人喘口氣的機(jī)會(huì)，LinkedBlockingQueue是一個(gè)基于鏈表的阻塞隊(duì)列，內(nèi)部是由節(jié)點(diǎn)Node構(gòu)成，每個(gè)被加入隊(duì)列的元素都會(huì)被封裝成下面的Node節(jié)點(diǎn)，并且節(jié)點(diǎn)中有指向下一個(gè)元素的指針：

static class Node<E> { 
    E item; 
    Node<E> next; 
    Node(E x) { item = x; } 
} 

LinkedBlockingQueue中的關(guān)鍵屬性有下面這些：

private final int capacity;//隊(duì)列容量 
private final AtomicInteger count = new AtomicInteger();//隊(duì)列中元素?cái)?shù)量 
transient Node<E> head;//頭節(jié)點(diǎn) 
private transient Node<E> last;//尾節(jié)點(diǎn) 
//出隊(duì)鎖 
private final ReentrantLock takeLock = new ReentrantLock(); 
//出隊(duì)的等待條件對(duì)象 
private final Condition notEmpty = takeLock.newCondition(); 
//入隊(duì)鎖 
private final ReentrantLock putLock = new ReentrantLock(); 
//入隊(duì)的等待條件對(duì)象 
private final Condition notFull = putLock.newCondition(); 

構(gòu)造函數(shù)分為指定隊(duì)列長(zhǎng)度和不指定隊(duì)列長(zhǎng)度兩種，不指定時(shí)隊(duì)列最大長(zhǎng)度是int的最大值。當(dāng)然了，你要是真存這么多的元素，很有可能會(huì)引起內(nèi)存溢出：

public LinkedBlockingQueue() { 
    this(Integer.MAX_VALUE); 
} 
public LinkedBlockingQueue(int capacity) { 
    if (capacity <= 0) throw new IllegalArgumentException(); 
    this.capacity = capacity; 
    last = head = new Node<E>(null); 
}  

還有另一種在初始化時(shí)就可以將集合作為參數(shù)傳入的構(gòu)造方法，實(shí)現(xiàn)非常好理解，只是循環(huán)調(diào)用了后面會(huì)講到的enqueue入隊(duì)方法，這里暫且略過(guò)。

在LinkedBlockingQueue中，隊(duì)列的頭結(jié)點(diǎn)head是不存元素的，它的item是null，next指向的元素才是真正的第一個(gè)元素，它也有兩個(gè)用于阻塞等待的Condition條件對(duì)象。與之前的ArrayBlockingQueue不同，這里出隊(duì)和入隊(duì)使用了不同的鎖takeLock和putLock。隊(duì)列的結(jié)構(gòu)是這樣的：

面試官：為什么要使用兩把鎖，之前ArrayBlockingQueue使用一把鎖，不是也可以保證線程的安全么?

Hydra：使用兩把鎖，可以保證元素的插入和刪除并不互斥，從而能夠同時(shí)進(jìn)行，達(dá)到提高吞吐量的的效果

面試官：嗯，那還是老規(guī)矩，先說(shuō)插入方法是怎么實(shí)現(xiàn)的吧

Hydra：這次就不提父類AbstractQueue的add方法了，反正它調(diào)用的也是子類的插入方法offer，我們就直接來(lái)看offer方法的源碼：

public boolean offer(E e) { 
    if (e == null) throw new NullPointerException(); 
    final AtomicInteger count = this.count;//隊(duì)列中元素個(gè)數(shù) 
    if (count.get() == capacity)//已滿 
        return false; 
    int c = -1; 
    Node<E> node = new Node<E>(e); 
    final ReentrantLock putLock = this.putLock; 
    putLock.lock(); 
    try { 
        //并發(fā)情況，再次判斷隊(duì)列是否已滿 
        if (count.get() < capacity) { 
            enqueue(node); 
            //注意這里獲取的是未添加元素前的對(duì)列長(zhǎng)度 
            c = count.getAndIncrement(); 
            if (c + 1 < capacity)//未滿 
                notFull.signal(); 
        } 
    } finally { 
        putLock.unlock(); 
    } 
    if (c == 0) 
        signalNotEmpty(); 
    return c >= 0; 
} 

offer方法中，首先判斷隊(duì)列是否已滿，未滿情況下將元素封裝成Node對(duì)象，嘗試獲取插入鎖，在獲取鎖后會(huì)再進(jìn)行一次隊(duì)列已滿判斷，如果已滿則直接釋放鎖。在持有鎖且隊(duì)列未滿的情況下，調(diào)用enqueue入隊(duì)方法。

enqueue方法的實(shí)現(xiàn)也非常的簡(jiǎn)單，將當(dāng)前尾節(jié)點(diǎn)的next指針指向新節(jié)點(diǎn)，再把last指向新節(jié)點(diǎn)：

private void enqueue(Node<E> node) { 
    last = last.next = node; 
} 

畫(huà)一張圖，方便你理解：

在完成入隊(duì)后，判斷隊(duì)列是否已滿，如果未滿則調(diào)用notFull.signal()，喚醒等待將元素插入隊(duì)列的線程。

面試官：我記得在ArrayBlockingQueue里插入元素后，是調(diào)用的notEmpty.signal()，怎么這里還不一樣了?

Hydra：說(shuō)到這，就不得不再提一下使用兩把鎖來(lái)分別控制插入和獲取元素的好處了。在ArrayBlockingQueue中，使用了同一把鎖對(duì)入隊(duì)和出隊(duì)進(jìn)行控制，那么如果在插入元素后再喚醒插入線程，那么很有可能等待獲取元素的線程就一直得不到喚醒，造成等待時(shí)間過(guò)長(zhǎng)。

而在LinkedBlockingQueue中，分別使用了入隊(duì)鎖putLock和出隊(duì)鎖takeLock，插入線程和獲取線程是不會(huì)互斥的。所以插入線程可以在這里不斷的喚醒其他的插入線程，而無(wú)需擔(dān)心是否會(huì)使獲取線程等待時(shí)間過(guò)長(zhǎng)，從而在一定程度上提高了吞吐量。當(dāng)然了，因?yàn)閛ffer方法是非阻塞的，并不會(huì)掛起阻塞線程，所以這里喚醒的是阻塞插入的put方法的線程。

面試官：那接著往下看，為什么要在c等于0的情況下才去喚醒notEmpty中的等待獲取元素的線程?

Hydra：其實(shí)獲取元素的方法和上面插入元素的方法是一個(gè)模式的，只要有一個(gè)獲取線程在執(zhí)行方法，那么就會(huì)不斷的通過(guò)notEmpty.signal()喚醒其他的獲取線程。只有當(dāng)c等于0時(shí)，才證明之前隊(duì)列中已經(jīng)沒(méi)有元素，這時(shí)候獲取線程才可能會(huì)被阻塞，在這個(gè)時(shí)候才需要被喚醒。上面的這些可以用一張圖來(lái)說(shuō)明：

由于我們之前說(shuō)過(guò)，隊(duì)列中的head節(jié)點(diǎn)可以認(rèn)為是不存儲(chǔ)數(shù)據(jù)的標(biāo)志性節(jié)點(diǎn)，所以可以簡(jiǎn)單的認(rèn)為出隊(duì)時(shí)直接取出第二個(gè)節(jié)點(diǎn)，當(dāng)然這個(gè)過(guò)程不是非常的嚴(yán)謹(jǐn)，我會(huì)在后面講解出隊(duì)的過(guò)程中再進(jìn)行補(bǔ)充說(shuō)明。

面試官：那么阻塞方法put和它有什么區(qū)別?

Hydra：put和offer方法整體思路一致，不同的是加鎖是使用的是可被中斷的方式，并且當(dāng)隊(duì)列中元素已滿時(shí)，將線程加入notFull等待隊(duì)列中進(jìn)行等待，代碼中體現(xiàn)在：

while (count.get() == capacity) { 
    notFull.await(); 
} 

這個(gè)過(guò)程體現(xiàn)在上面那張圖的notFull等待隊(duì)列中的元素上，就不重復(fù)說(shuō)明了。另外，和put方法比較類似的，還有一個(gè)攜帶等待時(shí)間參數(shù)的offer方法，可以進(jìn)行有限時(shí)間內(nèi)的阻塞添加，當(dāng)超時(shí)后放棄插入元素，我們只看和offer方法不同部分的代碼：

public boolean offer(E e, long timeout, TimeUnit unit){ 
    ... 
    long nanos = unit.toNanos(timeout);//轉(zhuǎn)換為納秒 
    ... 
    while (count.get() == capacity) { 
        if (nanos <= 0) 
            return false; 
        nanos = notFull.awaitNanos(nanos); 
    } 
    enqueue(new Node<E>(e));     
    ... 
} 

awaitNanos方法在await方法的基礎(chǔ)上，增加了超時(shí)跳出的機(jī)制，會(huì)在循環(huán)中計(jì)算是否到達(dá)預(yù)設(shè)的超時(shí)時(shí)間。如果在到達(dá)超時(shí)時(shí)間前被喚醒，那么會(huì)返回超時(shí)時(shí)間減去已經(jīng)消耗的時(shí)間。無(wú)論是被其他線程喚醒返回，還是到達(dá)指定的超時(shí)時(shí)間返回，只要方法返回值小于等于0，那么就認(rèn)為它已經(jīng)超時(shí)，最終直接返回false結(jié)束。

面試官：費(fèi)這么大頓功夫才把插入講明白，我先喝口水，你接著說(shuō)獲取元素方法

Hydra：……那先看非阻塞的poll方法

public E poll() { 
    final AtomicInteger count = this.count; 
    if (count.get() == 0)//隊(duì)列為空 
        return null; 
    E x = null; 
    int c = -1; 
    final ReentrantLock takeLock = this.takeLock; 
    takeLock.lock(); 
    try { 
        if (count.get() > 0) {//隊(duì)列非空 
            x = dequeue(); 
            //出隊(duì)前隊(duì)列長(zhǎng)隊(duì) 
            c = count.getAndDecrement(); 
            if (c > 1) 
                notEmpty.signal(); 
        } 
    } finally { 
        takeLock.unlock(); 
    } 
    if (c == capacity) 
        signalNotFull(); 
    return x; 
} 

出隊(duì)的邏輯和入隊(duì)的非常相似，當(dāng)隊(duì)列非空時(shí)就執(zhí)行dequeue進(jìn)行出隊(duì)操作，完成出隊(duì)后如果隊(duì)列仍然非空，那么喚醒等待隊(duì)列中掛起的獲取元素的線程。并且當(dāng)出隊(duì)前的元素?cái)?shù)量等于隊(duì)列長(zhǎng)度時(shí)，在出隊(duì)后喚醒等待隊(duì)列上的添加線程。

出隊(duì)方法dequeue的源碼如下：

private E dequeue() { 
    Node<E> h = head; 
    Node<E> first = h.next; 
    h.next = h; // help GC 
    head = first; 
    E x = first.item; 
    first.item = null; 
    return x; 
} 

之前提到過(guò)，頭節(jié)點(diǎn)head并不存儲(chǔ)數(shù)據(jù)，它的下一個(gè)節(jié)點(diǎn)才是真正意義上的第一個(gè)節(jié)點(diǎn)。在出隊(duì)操作中，先得到頭結(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)first節(jié)點(diǎn)，將當(dāng)前頭結(jié)點(diǎn)的next指針指向自己，代碼中有一個(gè)簡(jiǎn)單的注釋是help gc，個(gè)人理解這里是為了降低gc中的引用計(jì)數(shù)，方便它更早被回收。之后再將新的頭節(jié)點(diǎn)指向first，并返回清空為null前的內(nèi)容。使用圖來(lái)表示是這樣的：

面試官：(看看手表)take方法的整體邏輯也差不多，能簡(jiǎn)單概括一下嗎

Hydra：阻塞方法take方法和poll的思路基本一致，是一個(gè)可以被中斷的阻塞獲取方法，在隊(duì)列為空時(shí)，會(huì)掛起當(dāng)前線程，將它添加到條件對(duì)象notEmpty的等待隊(duì)列中，等待其他線程喚醒。

面試官：再給你一句話的時(shí)間，總結(jié)一下它和ArrayBlockingQueue的異同，我要下班回家了

Hydra：好吧，我總結(jié)一下，有下面幾點(diǎn)：

1、隊(duì)列長(zhǎng)度不同，ArrayBlockingQueue創(chuàng)建時(shí)需指定長(zhǎng)度并且不可修改，而LinkedBlockingQueue可以指定也可以不指定長(zhǎng)度

2、存儲(chǔ)方式不同，ArrayBlockingQueue使用數(shù)組，而LinkedBlockingQueue使用Node節(jié)點(diǎn)的鏈表

3、ArrayBlockingQueue使用一把鎖來(lái)控制元素的插入和移除，而LinkedBlockingQueue將入隊(duì)鎖和出隊(duì)鎖分離，提高了并發(fā)性能

4、ArrayBlockingQueue采用數(shù)組存儲(chǔ)元素，因此在插入和移除過(guò)程中不需要生成額外對(duì)象，LinkedBlockingQueue會(huì)生成新的Node節(jié)點(diǎn)，對(duì)gc會(huì)有影響