阻塞佇列BlockingQueue簡介

阻塞佇列BlockingQueue是JDK1.5併發新特性中的內容，阻塞佇列首先是一個佇列，同樣實現了Collection介面。阻塞佇列提供了可阻塞的put和take方法，以及支援定時的poll和offer方法。

阻塞佇列跟普通佇列相比，首頁它是執行緒安全的，另外還提供了兩個附加操作：當佇列為空時，從佇列中獲取元素的操作將被阻塞；當佇列填滿是，向佇列新增元素將被阻塞。這兩個附加操作分別由BlockingQueue提供的兩個take和put方法支援。如果佇列已經滿了，那麼put方法將被阻塞直到有空間可用；如果佇列為空，那麼take方法將被阻塞直到有元素可用。佇列可以是有界的也可以是無界的，無界佇列永遠不會充滿，因此在無界佇列上面put方法也永遠不會被阻塞。

BlockingQueue提供了4中型別的處理方法：

• 丟擲異常： 當佇列滿時，再向佇列中插入元素，則會丟擲IllegalStateException異常。當佇列空時，再向佇列中獲取元素，則會丟擲NoSuchElementException異常。
• 返回特殊值： 當佇列滿時，向佇列中新增元素，則返回false，否則返回true。當佇列為空時，向佇列中獲取元素，則返回null，否則返回元素。
• 一直阻塞： 當阻塞佇列滿時，如果生產者向佇列中插入元素，則佇列會一直阻塞當前執行緒，直到佇列可用或響應中斷退出。當阻塞佇列為空時，如果消費者執行緒向阻塞佇列中獲取資料，則佇列會一直阻塞當前執行緒，直到佇列空閒或響應中斷退出。
• 超時退出： 當佇列滿時，如果生產執行緒向佇列中新增元素，則佇列會阻塞生產執行緒一段時間，超過指定的時間則退出返回false。當佇列為空時，消費執行緒從佇列中移除元素，則佇列會阻塞一段時間，如果超過指定時間退出返回null。

Java提供的7個阻塞佇列

在多執行緒環境中，通過佇列可以很容易的實現資料共享。在基於佇列的生產者-消費者模型中，資料生產時，生產者就把資料放入佇列，當消費者準備使用資料時就從佇列中取出資料。生產者不需要知道消費者的標識或者數量，或者他們是唯一的生產者。同樣，消費者也不需要知道生產者來自何處。BlockingQqueue簡化了生產者-消費者的過程，它支援任意數量的生產者-消費者。一種最常見的生產者-消費者模式就是執行緒池與工作佇列的組合，在Executor執行框架中就體現了這種模式。

阻塞佇列BlockingQueue的成員介紹

ArrayBlockingQueue

ArrayBlockingQueue是一個基於陣列的阻塞佇列實現，內部維護了一個定長陣列，以便快取資料。一旦建立了這樣的快取區，就不能再增加其容量。試圖向已滿佇列中放入元素會導致放入操作受阻塞；試圖從空佇列中檢索元素將導致類似阻塞。ArrayBlockingQueue內部還儲存著兩個整形變數，分別標識著佇列的頭部和尾部在陣列中的位置。

• ArrayBlockingQueue(int capacity) 建立一個帶有給定的（固定）容量和預設訪問策略（非公平鎖）的 ArrayBlockingQueue。capacity是佇列容量。
• ArrayBlockingQueue(int capacity, boolean fair) 建立一個具有給定的（固定）容量和指定訪問策略的 ArrayBlockingQueue。fair訪問策略如果為 true，則按照 FIFO 順序訪問插入或移除時受阻塞執行緒的佇列，如果為 false，則訪問順序是不確定的。fair是“可重入的獨佔鎖(ReentrantLock)”的型別。fair為true，表示是公平鎖，fair為false，表示是非公平鎖。
• ArrayBlockingQueue(int capacity, boolean fair, Collectionc) 建立一個具有給定的（固定）容量和指定訪問策略的 ArrayBlockingQueue，它最初包含給定 collection 的元素，並以 collection 迭代器的遍歷順序新增元素。

由於ArrayBlockingQueue內部只維護一個ReentrantLock型別的lock鎖物件，所以在生成者-消費者模型中，並不能真正的實現並行，這一點不同於LinkedBlockingQueue，LinkedBlockingQueue內部維護了兩個鎖。事實上ArrayBlockingQueue完全可以採用分離鎖，從而實現生產者和消費者操作的完全並行執行。Doug Lea之所以沒這樣去做，也許是因為ArrayBlockingQueue的資料寫入和獲取操作已經足夠輕巧，以至於引入獨立的鎖機制，除了給程式碼帶來額外的複雜性外，其在效能上完全佔不到任何便宜。

ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在於，前者在插入或刪除元素時不會產生或銷燬任何額外的物件例項，而後者則會生成一個額外的Node物件。這在長時間內需要高效併發地處理大批量資料的系統中，其對於GC的影響還是存在一定的區別。而在建立ArrayBlockingQueue時，我們還可以控制物件的內部鎖是否採用公平鎖，預設採用非公平鎖。

LinkedBlockingQueue

LinkedBlockingQueue是一個單向連結串列實現的阻塞佇列，支援真正的並行操作，因為內部使用ReentrantLock實現插入鎖(putLock)和取出鎖(takeLock)，維護了兩個所物件。其內部也維持著一個數據緩衝佇列（該佇列由一個連結串列構成），當生產者往佇列中放入一個數據時，佇列會從生產者手中獲取資料，並快取在佇列內部，而生產者立即返回；只有當佇列緩衝區達到最大值快取容量時（LinkedBlockingQueue可以通過建構函式指定該值），才會阻塞生產者佇列，直到消費者從佇列中消費掉一份資料，生產者執行緒會被喚醒，反之對於消費者這端的處理也基於同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理併發資料，還因為其對於生產者端和消費者端分別採用了獨立的鎖來控制資料同步，這也意味著在高併發的情況下生產者和消費者可以並行地操作佇列中的資料，以此來提高整個佇列的併發效能。

在開發中新建LinkedBlockingQueue例項的時候，一般要指定其大小，如果沒有指定大小，大小預設是Integer.MAX_VALUE，這樣的話，如果生產者的速度一旦大於消費者的速度，也許還沒有等到佇列滿阻塞產生，系統記憶體就有可能已被消耗殆盡了。線上程池框架Executors中newSingleThreadExecutor和newFixedThreadPool方法內部維護的都是LinkedBlockingQueue。

PriorityBlockingQueue

PriorityBlockingQueue是一個按照優先順序排序的佇列，如果想要某個佇列不是按照FIFO的順序來處理元素，該佇列非常有用，內部維護一個堆的資料結構。PriorityBlockingQueue既可以根據元素的自然順序進行排序，如果元素實現了Comparable介面，也可以根據Comparator進行比較。該佇列看似有界佇列，實際上它會自動擴容，因此是無界佇列，因此在生產者-消費者模型中，生產者並不會真正的阻塞，而只會在沒有可消費的資料時，阻塞資料的消費者。因此使用的時候要特別注意，生產者生產資料的速度絕對不能快於消費者消費資料的速度，否則時間一長，會最終耗盡所有的可用堆記憶體空間。在實現PriorityBlockingQueue時，內部控制執行緒同步的鎖採用的是非公平鎖。

DelayQueue

DelayQueue是一個無界阻塞佇列，只有在延遲期滿時才能從中提取元素。該佇列的頭部 是延遲期滿後儲存時間最長的 Delayed 元素。如果延遲都還沒有期滿，則佇列沒有頭部，並且 poll 將返回null。當一個元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一個小於等於 0 的值時，將發生到期。即使無法使用 take 或 poll移除未到期的元素，也不會將這些元素作為正常元素對待。例如，size 方法同時返回到期和未到期元素的計數。此佇列不允許使用 null 元素。

DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個超時未響應的連線佇列。

SynchronousQueue

SynchronousQueue是這樣一種阻塞佇列，其中每個 put 必須等待一個take，反之亦然。同步佇列沒有任何內部容量，甚至連一個佇列的容量都沒有，它不會為佇列中的元素維護儲存空間。與其它佇列不同的是，它維護一組執行緒，這些執行緒在等待著元素加入或者移除佇列。不能在同步佇列上進行peek，因為僅在試圖要取得元素時，該元素才存在；除非另一個執行緒試圖移除某個元素，否則也不能（使用任何方法）新增元素；也不能迭代佇列，因為其中沒有元素可用於迭代。佇列的頭是嘗試新增到佇列中的首個已排隊執行緒元素； 如果沒有已排隊執行緒，則不新增元素並且頭為null。SynchronousQueue類似於無中介的直接交易，有點像原始社會中的生產者和消費者，生產者拿著產品去集市銷售給產品的最終消費者，而消費者必須親自去集市找到所要商品的直接生產者，如果一方沒有找到合適的目標，那麼對不起，大家都在集市等待。

SynchronousQueue的一個使用場景是線上程池裡。Executors.newCachedThreadPool()就使用了SynchronousQueue，這個執行緒池根據需要（新任務到來時）建立新的執行緒，如果有空閒執行緒則會重複使用，執行緒空閒了60秒後會被回收。

建立SynchronousQueue有兩種構造方法，一種時SynchronousQueue()，預設採用非公平的形式，從JDK1.6開始SynchronousQueue的實現採用了一種效能更好的無鎖演算法。競爭機制支援公平和非公平兩種：非公平競爭模式使用的資料結構是後進先出棧(LIFO Stack)；公平競爭模式則使用先進先出佇列（FIFO），效能上兩者是相當的，一般情況下，FIFO通常可以支援更大的吞吐量，但LIFO可以更大程度的保持執行緒的本地化。另外一種SynchronousQueue(boolean fair)，可以自己指定訪問方式是否採用公平方式。

LinkedTransferQueue

LinkedTransferQueue是JDK1.7中新引入的佇列，該佇列的實現基於CAS無鎖機制，它也是一個基於連結串列實現的無界佇列。相比前面佇列它多transfer和tryTransfer方法。

LinkedBlockingDeque

LinkedBlockingDeque一個基於已連結節點的、任選範圍的阻塞雙端佇列。可選的容量範圍構造方法引數是一種防止過度膨脹的方式。如果未指定容量，那麼容量將等於 Integer.MAX_VALUE。只要插入元素不會使雙端佇列超出容量，每次插入後都將動態地建立連結節點。要想支援阻塞功能，佇列的容量一定是固定的，否則無法在入隊的時候掛起執行緒。也就是capacity是final型別的。

阻塞佇列示例

這是一個使用LinkedBlockedQueue設計實現的簡單的生產者-消費者模式。

public class Producer implements Runnable {
 
 private volatile boolean isRunning = true;
 private BlockingQueue<String> queue;
 private static final int DEFAULT_SLEEP = 1000;
 private static AtomicInteger count = new AtomicInteger();
 
 public Producer(BlockingQueue<String> queue) {
 this.queue = queue;
 }
 
 @Override
 public void run() {
 String data = null;
 Random r = new Random();
 
 System.out.println("啟動生產者執行緒！");
 try {
 while (isRunning) {
 Thread.sleep(r.nextInt(DEFAULT_SLEEP));
 data = "data:" + count.incrementAndGet();
 queue.put(data);
 System.out.println("將資料：" + data + "放入佇列...");
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 Thread.currentThread().interrupt();
 } finally {
 System.out.println("退出生產者執行緒！");
 }
 }
 
 public void stop() {
 isRunning = false;
 }
 
}
public class Consumer implements Runnable {
 
 private volatile boolean isRunning = true;
 private BlockingQueue<String> queue;
 private static final int DEFAULT_SLEEP = 1000;
 
 public Consumer(BlockingQueue<String> queue) {
 this.queue = queue;
 }
 
 @Override
 public void run() {
 System.out.println("啟動消費者執行緒！");
 Random r = new Random();
 try {
 while (isRunning) {
 String data = queue.take();
 if (null != data) {
 System.out.println("正在消費：" + data);
 Thread.sleep(r.nextInt(DEFAULT_SLEEP));
 }
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 Thread.currentThread().interrupt();
 } finally {
 System.out.println("退出消費者執行緒！");
 }
 }
 
 public void stop() {
 isRunning = false;
 }
 
}
public class MainTest {
 
 public static void main(String[] args) throws InterruptedException {
 // 宣告一個容量為10的快取佇列
 BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
 
 Producer producer1 = new Producer(queue);
 Producer producer2 = new Producer(queue);
 Producer producer3 = new Producer(queue);
 Consumer consumer = new Consumer(queue);
 
 // 藉助Executors
 ExecutorService service = Executors.newCachedThreadPool();
 // 啟動執行緒
 service.execute(producer1);
 service.execute(producer2);
 service.execute(producer3);
 service.execute(consumer);
 
 // 執行20s
 Thread.sleep(20 * 1000);
 producer1.stop();
 producer2.stop();
 producer3.stop();
 
 Thread.sleep(2 * 1000);
 consumer.stop();
 // 退出Executor
 service.shutdown();
 }
 
}

如果不使用阻塞佇列，使用Object.wait()和Object.notify()非阻塞佇列實現生產者-消費者模式，生產者執行緒在緩衝區為滿的時候，消費者在緩衝區為空的時候，都應該暫停執行。然後用notify 和notifyAll通知等待中的執行緒重新開始執行。