原文地址：http://benjaminwhx.com/2018/05/11/%E3%80%90%E7%BB%86%E8%B0%88Java%E5%B9%B6%E5%8F%91%E3%80%91%E8%B0%88%E8%B0%88LinkedBlockingQueue/

在集合框架里，想必大家都用過ArrayList和LinkedList，也經(jīng)常在面試中問到他們之間的區(qū)別。ArrayList和ArrayBlockingQueue一樣，內(nèi)部基于數(shù)組來存放元素，而LinkedBlockingQueue則和LinkedList一樣，內(nèi)部基于鏈表來存放元素。

LinkedBlockingQueue實(shí)現(xiàn)了BlockingQueue接口，這里放一張類的繼承關(guān)系圖（圖片來自之前的文章：說說隊(duì)列Queue）

LinkedBlockingQueue不同于ArrayBlockingQueue，它如果不指定容量，默認(rèn)為Integer.MAX_VALUE，也就是無界隊(duì)列。所以為了避免隊(duì)列過大造成機(jī)器負(fù)載或者內(nèi)存爆滿的情況出現(xiàn)，我們在使用的時(shí)候建議手動(dòng)傳一個(gè)隊(duì)列的大小。

2、源碼分析

2.1 屬性

/*** 節(jié)點(diǎn)類，用于存儲(chǔ)數(shù)據(jù)*/ static class Node<E> {E item;Node<E> next;Node(E x) { item = x; } }/** 阻塞隊(duì)列的大小，默認(rèn)為Integer.MAX_VALUE */ private final int capacity;/** 當(dāng)前阻塞隊(duì)列中的元素個(gè)數(shù) */ private final AtomicInteger count = new AtomicInteger();/*** 阻塞隊(duì)列的頭結(jié)點(diǎn)*/ transient Node<E> head;/*** 阻塞隊(duì)列的尾節(jié)點(diǎn)*/ private transient Node<E> last;/** 獲取并移除元素時(shí)使用的鎖，如take, poll, etc */ private final ReentrantLock takeLock = new ReentrantLock();/** notEmpty條件對象，當(dāng)隊(duì)列沒有數(shù)據(jù)時(shí)用于掛起執(zhí)行刪除的線程 */ private final Condition notEmpty = takeLock.newCondition();/** 添加元素時(shí)使用的鎖如 put, offer, etc */ private final ReentrantLock putLock = new ReentrantLock();/** notFull條件對象，當(dāng)隊(duì)列數(shù)據(jù)已滿時(shí)用于掛起執(zhí)行添加的線程 */ private final Condition notFull = putLock.newCondition();

從上面的屬性我們知道，每個(gè)添加到LinkedBlockingQueue隊(duì)列中的數(shù)據(jù)都將被封裝成Node節(jié)點(diǎn)，添加的鏈表隊(duì)列中，其中head和last分別指向隊(duì)列的頭結(jié)點(diǎn)和尾結(jié)點(diǎn)。與ArrayBlockingQueue不同的是，LinkedBlockingQueue內(nèi)部分別使用了takeLock 和 putLock 對并發(fā)進(jìn)行控制，也就是說，添加和刪除操作并不是互斥操作，可以同時(shí)進(jìn)行，這樣也就可以大大提高吞吐量。

這里如果不指定隊(duì)列的容量大小，也就是使用默認(rèn)的Integer.MAX_VALUE，如果存在添加速度大于刪除速度時(shí)候，有可能會(huì)內(nèi)存溢出，這點(diǎn)在使用前希望慎重考慮。

另外，LinkedBlockingQueue對每一個(gè)lock鎖都提供了一個(gè)Condition用來掛起和喚醒其他線程。

構(gòu)造函數(shù)

public LinkedBlockingQueue() {// 默認(rèn)大小為Integer.MAX_VALUEthis(Integer.MAX_VALUE); }public LinkedBlockingQueue(int capacity) {if (capacity <= 0) throw new IllegalArgumentException();this.capacity = capacity;last = head = new Node<E>(null); }public LinkedBlockingQueue(Collection<? extends E> c) {this(Integer.MAX_VALUE);final ReentrantLock putLock = this.putLock;putLock.lock();try {int n = 0;for (E e : c) {if (e == null)throw new NullPointerException();if (n == capacity)throw new IllegalStateException("Queue full");enqueue(new Node<E>(e));++n;}count.set(n);} finally {putLock.unlock();} }

默認(rèn)的構(gòu)造函數(shù)和最后一個(gè)構(gòu)造函數(shù)創(chuàng)建的隊(duì)列大小都為Integer.MAX_VALUE，只有第二個(gè)構(gòu)造函數(shù)用戶可以指定隊(duì)列的大小。第二個(gè)構(gòu)造函數(shù)最后初始化了last和head節(jié)點(diǎn)，讓它們都指向了一個(gè)元素為null的節(jié)點(diǎn)。

方法

同樣，LinkedBlockingQueue也有著和ArrayBlockingQueue一樣的方法，我們先來看看入隊(duì)列的方法。

2.3.1、入隊(duì)方法

LinkedBlockingQueue提供了多種入隊(duì)操作的實(shí)現(xiàn)來滿足不同情況下的需求，入隊(duì)操作有如下幾種：

• void put(E e)；
• boolean offer(E e)；
• boolean offer(E e, long timeout, TimeUnit unit)。

put(E e)

public void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();int c = -1;Node<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;// 獲取鎖中斷 putLock.lockInterruptibly();try {//判斷隊(duì)列是否已滿，如果已滿阻塞等待while (count.get() == capacity) {notFull.await();}// 把node放入隊(duì)列中 enqueue(node);c = count.getAndIncrement();// 再次判斷隊(duì)列是否有可用空間，如果有喚醒下一個(gè)線程進(jìn)行添加操作if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}// 如果隊(duì)列中有一條數(shù)據(jù)，喚醒消費(fèi)線程進(jìn)行消費(fèi)if (c == 0)signalNotEmpty(); }

小結(jié)put方法來看，它總共做了以下情況的考慮：

• 隊(duì)列已滿，阻塞等待。
• 隊(duì)列未滿，創(chuàng)建一個(gè)node節(jié)點(diǎn)放入隊(duì)列中，如果放完以后隊(duì)列還有剩余空間，繼續(xù)喚醒下一個(gè)添加線程進(jìn)行添加。如果放之前隊(duì)列中沒有元素，放完以后要喚醒消費(fèi)線程進(jìn)行消費(fèi)。

offer(E e)

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<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;putLock.lock();try {// 隊(duì)列有可用空間，放入node節(jié)點(diǎn)，判斷放入元素后是否還有可用空間，// 如果有，喚醒下一個(gè)添加線程進(jìn)行添加操作。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僅僅對put方法改動(dòng)了一點(diǎn)點(diǎn)，當(dāng)隊(duì)列沒有可用元素的時(shí)候，不同于put方法的阻塞等待，offer方法直接方法false。

offer(E e, long timeout, TimeUnit unit)

public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {if (e == null) throw new NullPointerException();long nanos = unit.toNanos(timeout);int c = -1;final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;putLock.lockInterruptibly();try {// 等待超時(shí)時(shí)間nanos，超時(shí)時(shí)間到了返回falsewhile (count.get() == capacity) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}enqueue(new Node<E>(e));c = count.getAndIncrement();if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}if (c == 0)signalNotEmpty();return true; }

該方法只是對offer方法進(jìn)行了阻塞超時(shí)處理，使用了Condition的awaitNanos來進(jìn)行超時(shí)等待，這里為什么要用while循環(huán)？因?yàn)閍waitNanos方法是可中斷的，為了防止在等待過程中線程被中斷，這里使用while循環(huán)進(jìn)行等待過程中中斷的處理，繼續(xù)等待剩下需等待的時(shí)間。

轉(zhuǎn)載于:https://www.cnblogs.com/liyongliang/p/10697876.html

總結(jié)

以上是生活随笔為你收集整理的Java中的阻塞队列-LinkedBlockingQueue（二）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。