別眨眼網

面試官：好了，聊完了

ArrayBlockingQueue

，我們接著說說

LinkedBlockingQueue

吧

Hydra：還真是不給人喘口氣的機會，

LinkedBlockingQueue

是一個基於連結串列的阻塞佇列，內部是由節點

Node

構成，每個被加入佇列的元素都會被封裝成下面的

Node

節點，並且節點中有指向下一個元素的指標：

static class Node {    E item；    Node next；    Node（E x） { item = x； }}

LinkedBlockingQueue

中的關鍵屬性有下面這些：

private final int capacity；//佇列容量private final AtomicInteger count = new AtomicInteger（）；//佇列中元素數量transient Node head；//頭節點private transient Node last；//尾節點//出隊鎖private final ReentrantLock takeLock = new ReentrantLock（）；//出隊的等待條件物件private final Condition notEmpty = takeLock。newCondition（）；//入隊鎖private final ReentrantLock putLock = new ReentrantLock（）；//入隊的等待條件物件private final Condition notFull = putLock。newCondition（）；

建構函式分為指定佇列長度和不指定佇列長度兩種，不指定時佇列最大長度是

int

的最大值。當然了，你要是真存這麼多的元素，很有可能會引起記憶體溢位：

public LinkedBlockingQueue（） {    this（Integer。MAX_VALUE）；}public LinkedBlockingQueue（int capacity） {    if （capacity <= 0） throw new IllegalArgumentException（）；    this。capacity = capacity；    last = head = new Node（null）；}

還有另一種在初始化時就可以將集合作為引數傳入的構造方法，實現非常好理解，只是迴圈呼叫了後面會講到的

enqueue

入隊方法，這裡暫且略過。

在

LinkedBlockingQueue

中，佇列的頭節點

head

是不存元素的，它的

item

是

null

，

next

指向的元素才是真正的第一個元素，它也有兩個用於阻塞等待的

Condition

條件物件。與之前的

ArrayBlockingQueue

不同，這裡出隊和入隊使用了不同的鎖

takeLock

和

putLock

。佇列的結構是這樣的：

面試官：為什麼要使用兩把鎖，之前

ArrayBlockingQueue

使用一把鎖，不是也可以保證執行緒的安全麼？

Hydra：使用兩把鎖，可以保證元素的插入和刪除並不互斥，從而能夠同時進行，達到提高吞吐量的的效果

面試官：嗯，那還是老規矩，先說插入方法是怎麼實現的吧

Hydra：這次就不提父類

AbstractQueue

的

add

方法了，反正它呼叫的也是子類的插入方法

offer

，我們就直接來看

offer

方法的原始碼：

public boolean offer（E e） {    if （e == null） throw new NullPointerException（）；    final AtomicInteger count = this。count；//佇列中元素個數    if （count。get（） == capacity）//已滿        return false；    int c = -1；    Node node = new Node（e）；    final ReentrantLock putLock = this。putLock；    putLock。lock（）；    try {        //併發情況，再次判斷佇列是否已滿        if （count。get（） < capacity） {            enqueue（node）；            //注意這裡獲取的是未新增元素前的對列長度            c = count。getAndIncrement（）；            if （c + 1 < capacity）//未滿                notFull。signal（）；        }    } finally {        putLock。unlock（）；    }    if （c == 0）        signalNotEmpty（）；    return c >= 0；}

offer

方法中，首先判斷佇列是否已滿，未滿情況下將元素封裝成

Node

物件，嘗試獲取插入鎖，在獲取鎖後會再進行一次佇列已滿判斷，如果已滿則直接釋放鎖。在持有鎖且佇列未滿的情況下，呼叫

enqueue

入隊方法。

enqueue

方法的實現也非常的簡單，將當前尾節點的

next

指標指向新節點，再把

last

指向新節點：

private void enqueue（Node node） {    last = last。next = node；}

畫一張圖，方便你理解：

在完成入隊後，判斷佇列是否已滿，如果未滿則呼叫

notFull。signal（）

，喚醒等待將元素插入佇列的執行緒。

面試官：我記得在

ArrayBlockingQueue

裡插入元素後，是呼叫的

notEmpty。signal（）

，怎麼這裡還不一樣了？

Hydra：說到這，就不得不再提一下使用兩把鎖來分別控制插入和獲取元素的好處了。在

ArrayBlockingQueue

中，使用了同一把鎖對入隊和出隊進行控制，那麼如果在插入元素後再喚醒插入執行緒，那麼很有可能等待獲取元素的執行緒就一直得不到喚醒，造成等待時間過長。

而在

LinkedBlockingQueue

中，分別使用了入隊鎖

putLock

和出隊鎖

takeLock

，插入執行緒和獲取執行緒是不會互斥的。所以插入執行緒可以在這裡不斷的喚醒其他的插入執行緒，而無需擔心是否會使獲取執行緒等待時間過長，從而在一定程度上提高了吞吐量。當然了，因為

offer

方法是非阻塞的，並不會掛起阻塞執行緒，所以這裡喚醒的是阻塞插入的

put

方法的執行緒。

面試官：那接著往下看，為什麼要在

c

等於0的情況下才去喚醒

notEmpty

中的等待獲取元素的執行緒？

Hydra：其實獲取元素的方法和上面插入元素的方法是一個模式的，只要有一個獲取執行緒在執行方法，那麼就會不斷的透過

notEmpty。signal（）

喚醒其他的獲取執行緒。只有當

c

等於0時，才證明之前佇列中已經沒有元素，這時候獲取執行緒才可能會被阻塞，在這個時候才需要被喚醒。上面的這些可以用一張圖來說明：

由於我們之前說過，佇列中的

head

節點可以認為是不儲存資料的標誌性節點，所以可以簡單的認為出隊時直接取出第二個節點，當然這個過程不是非常的嚴謹，我會在後面講解出隊的過程中再進行補充說明。

面試官：那麼阻塞方法

put

和它有什麼區別？

Hydra：

put

和

offer

方法整體思路一致，不同的是加鎖是使用的是可被中斷的方式，並且當佇列中元素已滿時，將執行緒加入

notFull

等待佇列中進行等待，程式碼中體現在：

while （count。get（） == capacity） {    notFull。await（）；}

這個過程體現在上面那張圖的

notFull

等待佇列中的元素上，就不重複說明了。另外，和

put

方法比較類似的，還有一個攜帶等待時間引數的

offer

方法，可以進行有限時間內的阻塞新增，當超時後放棄插入元素，我們只看和

offer

方法不同部分的程式碼：

public boolean offer（E e， long timeout， TimeUnit unit）{    。。。    long nanos = unit。toNanos（timeout）；//轉換為納秒    。。。    while （count。get（） == capacity） {        if （nanos <= 0）            return false；        nanos = notFull。awaitNanos（nanos）；    }    enqueue（new Node（e））；        。。。}

awaitNanos

方法在

await

方法的基礎上，增加了超時跳出的機制，會在迴圈中計算是否到達預設的超時時間。如果在到達超時時間前被喚醒，那麼會返回超時時間減去已經消耗的時間。無論是被其他執行緒喚醒返回，還是到達指定的超時時間返回，只要方法返回值小於等於0，那麼就認為它已經超時，最終直接返回

false

結束。

面試官：費這麼大頓功夫才把插入講明白，我先喝口水，你接著說獲取元素方法

Hydra：……那先看非阻塞的

poll

方法

public E poll（） {    final AtomicInteger count = this。count；    if （count。get（） == 0）//佇列為空        return null；    E x = null；    int c = -1；    final ReentrantLock takeLock = this。takeLock；    takeLock。lock（）；    try {        if （count。get（） > 0） {//佇列非空            x = dequeue（）；            //出隊前佇列長隊            c = count。getAndDecrement（）；            if （c > 1）                notEmpty。signal（）；        }    } finally {        takeLock。unlock（）；    }    if （c == capacity）        signalNotFull（）；    return x；}

出隊的邏輯和入隊的非常相似，當佇列非空時就執行

dequeue

進行出隊操作，完成出隊後如果佇列仍然非空，那麼喚醒等待佇列中掛起的獲取元素的執行緒。並且當出隊前的元素數量等於佇列長度時，在出隊後喚醒等待佇列上的新增執行緒。

出隊方法

dequeue

的原始碼如下：

private E dequeue（） {    Node h = head；    Node first = h。next；    h。next = h； // help GC    head = first；    E x = first。item；    first。item = null；    return x；}

之前提到過，頭節點

head

並不儲存資料，它的下一個節點才是真正意義上的第一個節點。在出隊操作中，先得到頭節點的下一個節點

first

節點，將當前頭節點的

next

指標指向自己，程式碼中有一個簡單的註釋是

help gc

，個人理解這裡是為了降低

gc

中的引用計數，方便它更早被回收。之後再將新的頭節點指向

first

，並返回清空為

null

前的內容。使用圖來表示是這樣的：

面試官：（看看手錶）

take

方法的整體邏輯也差不多，能簡單概括一下嗎

Hydra：阻塞方法

take

方法和

poll

的思路基本一致，是一個可以被中斷的阻塞獲取方法，在佇列為空時，會掛起當前執行緒，將它新增到條件物件

notEmpty

的等待佇列中，等待其他執行緒喚醒。

面試官：再給你一句話的時間，總結一下它和

ArrayBlockingQueue

的異同，我要下班回家了

Hydra：好吧，我總結一下，有下面幾點：

佇列長度不同，

ArrayBlockingQueue

建立時需指定長度並且不可修改，而

LinkedBlockingQueue

可以指定也可以不指定長度

儲存方式不同，

ArrayBlockingQueue

使用陣列，而

LinkedBlockingQueue

使用

Node

節點的連結串列

ArrayBlockingQueue

使用一把鎖來控制元素的插入和移除，而

LinkedBlockingQueue

將入隊鎖和出隊鎖分離，提高了併發效能

ArrayBlockingQueue

採用陣列儲存元素，因此在插入和移除過程中不需要生成額外物件，

LinkedBlockingQueue

會生成新的

Node

節點，對

gc

會有影響

面試官：明天上午9點，老地方，我們再聊聊別的

Hydra：……

如果文章對您有所幫助，歡迎關注公眾號

碼農參上