在JAVA的Concurrent包中，BlockingQueue很好的解決了多線程中，如何高效安全“傳輸”數據的問題。通過這些高效並且線程安全的隊列類，為我們快速搭建高質量的多線程程序帶來極大的便利。本文詳細介紹了BlockingQueue家庭中的所有成員，包括他們各自的功能以及常見使用場景。
一、認識BlockingQueue
阻塞隊列，顧名思義，首先它是一個隊列，而一個隊列在數據結構中所起的作用大致如下圖所示：
從上圖我們可以很清楚看到，通過一個共享的隊列，可以使得數據由隊列的一端輸入，從另外一端輸出；
常用的隊列主要有以下兩種：（當然通過不同的實現方式，還可以延伸出很多不同類型的隊列，DelayQueue就是其中的一種）
先進先出（FIFO）：先插入的隊列的元素也最先出隊列，類似於排隊的功能。從某種程度上來說這種隊列也體現了一種公平性。
后進先出（LIFO）：后插入隊列的元素最先出隊列，這種隊列優先處理最近發生的事件。
多線程環境中，通過隊列可以很容易實現數據共享，比如經典的“生產者”和“消費者”模型中，通過隊列可以很便利地實現兩者之間的數據共享。假設我們有若干生產者線程，另外又有若干個消費者線程。如果生產者線程需要把准備好的數據共享給消費者線程，利用隊列的方式來傳遞數據，就可以很方便地解決他們之間的數據共享問題。但如果生產者和消費者在某個時間段內，萬一發生數據處理速度不匹配的情況呢？理想情況下，如果生產者產出數據的速度大於消費者消費的速度，並且當生產出來的數據累積到一定程度的時候，那么生產者必須暫停等待一下（阻塞生產者線程），以便等待消費者線程把累積的數據處理完畢，反之亦然。然而，在concurrent包發布以前，在多線程環境下，我們每個程序員都必須去自己控制這些細節，尤其還要兼顧效率和線程安全，而這會給我們的程序帶來不小的復雜度。好在此時，強大的concurrent包橫空出世了，而他也給我們帶來了強大BlockingQueue。（在多線程領域：所謂阻塞，在某些情況下會掛起線程（即阻塞），一旦條件滿足，被掛起的線程又會自動被喚醒）下面兩幅圖演示了BlockingQueue的兩個常見阻塞場景：
如上圖所示：當隊列中沒有數據的情況下，消費者端的所有線程都會被自動阻塞（掛起），直到有數據放入隊列。
如上圖所示：當隊列中填滿數據的情況下，生產者端的所有線程都會被自動阻塞（掛起），直到隊列中有空的位置，線程被自動喚醒。
這也是我們在多線程環境下，為什么需要BlockingQueue的原因。作為BlockingQueue的使用者，我們再也不需要關心什么時候需要阻塞線程，什么時候需要喚醒線程，因為這一切BlockingQueue都給你一手包辦了。既然BlockingQueue如此神通廣大，讓我們一起來見識下它的常用方法：
二、BlockingQueue定義的常用方法
|
|
拋出異常 |
特殊值 |
阻塞 |
超時 |
| 插入 |
add(e) |
offer(e) |
put(e) |
offer(e,time,unit) |
| 移除 |
remove() |
poll() |
take() |
poll(time,unit) |
| 檢查 |
element() |
peek()/isEmpty() |
無 |
無 |
其中： null被用作指示poll操作失敗的警戒值。
1) 插入對象：
BlockingQueue不接受null對象。試圖add、put或offer一個null對象時，拋出NullPointerException。
add(object):把object加到BlockingQueue里,如果BlockingQueue可以容納,則返回true,否則拋出異常。
offer(object):把object加到BlockingQueue里,如果BlockingQueue可以容納,則返回true,否則返回false，本方法不阻塞當前執行方法的線程，offer(E o, long timeout, TimeUnit unit),可以設定等待的時間，如果在指定的時間內，不能加入BlockingQueue，則返回false。 put(object):把object加到BlockingQueue里,如果BlockQueue沒有空間,則調用此方法的線程被阻塞,直到BlockingQueue里面有空間再繼續加到BlockingQueue。
2) 獲取對象：
poll(time):取走BlockingQueue里排在首位的對象,若不能立即取出,則可以等time參數規定的時間,取不到時返回null;poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，如果在指定時間內，隊列一旦有數據可取，則立即返回隊列中的數據。否則超過時間還沒有數據可取，返回null。
take():取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻塞去數據線程進入等待狀態直到BlockingQueue有新加入的對象被取走;
drainTo():一次性從BlockingQueue獲取所有可用的數據對象，也可以一次性從BlockingQueue獲取指定個數的數據對象），通過該方法，可以提升獲取數據效率；不需要多次分批加鎖或釋放鎖。
3) 檢查對象
element() 獲取BlockingQueue里排在首位的對象並不從隊列中移除。如果隊列沒有對象，拋出異常。
peek()獲取BlockingQueue里排在首位的對象並不從隊列中移除。如果隊列沒有對象，返回null對象。
isEmpty() 判斷BlockingQueue里是否有對象，如果為空，返回true，如果不為空，返回false。
三、常見BlockingQueue
在了解了BlockingQueue的基本功能后，讓我們來看看BlockingQueue家庭大致有哪些成員？
1. ArrayBlockingQueue
基於數組的阻塞隊列實現，在ArrayBlockingQueue內部，維護了一個定長數組，以便緩存隊列中的數據對象，這是一個常用的阻塞隊列，除了一個定長數組外，ArrayBlockingQueue內部還保存着兩個整形變量，分別標識着隊列的頭部和尾部在數組中的位置。ArrayBlockingQueue在生產者放入數據和消費者獲取數據，都是共用同一個鎖對象，由此也意味着兩者無法真正並行運行，這點尤其不同於LinkedBlockingQueue；按照實現原理來分析，ArrayBlockingQueue完全可以采用分離鎖，從而實現生產者和消費者操作的完全並行運行。Doug Lea之所以沒這樣去做，也許是因為ArrayBlockingQueue的數據寫入和獲取操作已經足夠輕巧，以至於引入獨立的鎖機制，除了給代碼帶來額外的復雜性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在於，前者在插入或刪除元素時不會產生或銷毀任何額外的對象實例，而后者則會生成一個額外的Node對象。這在長時間內需要高效並發地處理大批量數據的系統中，其對於GC的影響還是存在一定的區別。而在創建ArrayBlockingQueue時，我們還可以控制對象的內部鎖是否采用公平鎖，默認采用非公平鎖。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class BlockingQueueTest {
public static void main(String[] args) throws InterruptedException {
// 聲明一個容量為10的緩存隊列
BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
queue.add("sss");
//測試添加null對象
try{
queue.add(null);
}
catch(Exception ee){
ee.printStackTrace();
}
try{
queue.offer(null);
}
catch(Exception ee){
ee.printStackTrace();
}
try{
queue.put(null);
}
catch(Exception ee){
ee.printStackTrace();
}
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);
// 執行10s
Thread.sleep(10 * 1000);
producer1.stop();
producer2.stop();
producer3.stop();
Thread.sleep(2000);
// 退出Executor
service.shutdown();
}
}import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* 消費者線程
*/
public class Consumer implements Runnable {
public Consumer(BlockingQueue<String> queue) {
this.queue = queue;
}
public void run() {
System.out.println("啟動消費者線程！");
Random r = new Random();
boolean isRunning = true;
try {
while (isRunning) {
System.out.println("正從隊列獲取數據...");
String data = queue.poll(2, TimeUnit.SECONDS);
if (null != data) {
System.out.println("拿到數據：" + data);
System.out.println("正在消費數據：" + data);
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
} else {
// 超過2s還沒數據，認為所有生產線程都已經退出，自動退出消費線程。
isRunning = false;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出消費者線程！");
}
}
private BlockingQueue<String> queue;
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 生產者線程
*
* @author jackyuj
*/
public class Producer implements Runnable {
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
String data = null;
Random r = new Random();
System.out.println("啟動生產者線程！");
try {
while (isRunning) {
System.out.println("正在生產數據...");
Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
data = "data:" + count.incrementAndGet();
System.out.println("將數據：" + data + "放入隊列...");
if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
System.out.println("放入數據失敗：" + data);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
} finally {
System.out.println("退出生產者線程！");
}
}
public void stop() {
isRunning = false;
}
private volatile boolean isRunning = true;
private BlockingQueue queue;
private static AtomicInteger count = new AtomicInteger();
private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}
2. LinkedBlockingQueue
基於鏈表的阻塞隊列，同ArrayListBlockingQueue類似，其內部也維持着一個數據緩沖隊列（該隊列由一個鏈表構成），當生產者往隊列中放入一個數據時，隊列會從生產者手中獲取數據，並緩存在隊列內部，而生產者立即返回；只有當隊列緩沖區達到最大值緩存容量時（LinkedBlockingQueue可以通過構造函數指定該值），才會阻塞生產者隊列，直到消費者從隊列中消費掉一份數據，生產者線程會被喚醒，反之對於消費者這端的處理也基於同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理並發數據，還因為其對於生產者端和消費者端分別采用了獨立的鎖來控制數據同步，這也意味着在高並發的情況下生產者和消費者可以並行地操作隊列中的數據，以此來提高整個隊列的並發性能。
作為開發者，我們需要注意的是，如果構造一個LinkedBlockingQueue對象，而沒有指定其容量大小，LinkedBlockingQueue會默認一個類似無限大小的容量（Integer.MAX_VALUE），這樣的話，如果生產者的速度一旦大於消費者的速度，也許還沒有等到隊列滿阻塞產生，系統內存就有可能已被消耗殆盡了。
ArrayBlockingQueue和LinkedBlockingQueue是兩個最普通也是最常用的阻塞隊列，一般情況下，在處理多線程間的生產者消費者問題，使用這兩個類足以。
3. DelayQueue
DelayQueue中的元素只有當其指定的延遲時間到了，才能夠從隊列中獲取到該元素。DelayQueue是一個沒有大小限制的隊列，因此往隊列中插入數據的操作（生產者）永遠不會被阻塞，而只有獲取數據的操作（消費者）才會被阻塞。
使用場景：DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個超時未響應的連接隊列。
4. PriorityBlockingQueue
基於優先級的阻塞隊列（優先級的判斷通過構造函數傳入的Compator對象來決定），但需要注意的是PriorityBlockingQueue並不會阻塞數據生產者，而只會在沒有可消費的數據時，阻塞數據的消費者。因此使用的時候要特別注意，生產者生產數據的速度絕對不能快於消費者消費數據的速度，否則時間一長，會最終耗盡所有的可用堆內存空間。在實現PriorityBlockingQueue時，內部控制線程同步的鎖采用的是公平鎖。
PriorityBlockingQueue里面存儲的對象必須是實現Comparable接口。隊列通過這個接口的compare方法確定對象的priority。
5. SynchronousQueue
一種無緩沖的等待隊列，類似於無中介的直接交易，有點像原始社會中的生產者和消費者，生產者拿着產品去集市銷售給產品的最終消費者，而消費者必須親自去集市找到所要商品的直接生產者，如果一方沒有找到合適的目標，那么對不起，大家都在集市等待。相對於有緩沖的BlockingQueue來說，少了一個中間經銷商的環節（緩沖區），如果有經銷商，生產者直接把產品批發給經銷商，而無需在意經銷商最終會將這些產品賣給那些消費者，由於經銷商可以庫存一部分商品，因此相對於直接交易模式，總體來說采用中間經銷商的模式會吞吐量高一些（可以批量買賣）；但另一方面，又因為經銷商的引入，使得產品從生產者到消費者中間增加了額外的交易環節，單個產品的及時響應性能可能會降低。
聲明一個SynchronousQueue有兩種不同的方式，它們之間有着不太一樣的行為。公平模式和非公平模式的區別:
如果采用公平模式：SynchronousQueue會采用公平鎖，並配合一個FIFO隊列來阻塞多余的生產者和消費者，從而體系整體的公平策略；
但如果是非公平模式（SynchronousQueue默認）：SynchronousQueue采用非公平鎖，同時配合一個LIFO隊列來管理多余的生產者和消費者，而后一種模式，如果生產者和消費者的處理速度有差距，則很容易出現飢渴的情況，即可能有某些生產者或者是消費者的數據永遠都得不到處理。
6.小結
BlockingQueue不光實現了一個完整隊列所具有的基本功能，同時在多線程環境下，他還自動管理了多線間的自動等待於喚醒功能，從而使得程序員可以忽略這些細節，關注更高級的功能。
ArrayBlockingQueue:規定大小的BlockingQueue,其構造函數必須帶一個int參數來指明其大小.其所含的對象是以FIFO(先入先出)順序排序的；
LinkedBlockingQueue:大小不定的BlockingQueue,若其構造函數帶一個規定大小的參數,生成的BlockingQueue有大小限制,若不帶大小參數,所生成的BlockingQueue的大小由Integer.MAX_VALUE來決定.其所含的對象是以FIFO(先入先出)順序排序；
PriorityBlockingQueue:類似於LinkedBlockQueue,但其所含對象的排序不是FIFO,而是依據對象的自然排序順序或者是構造函數的Comparator決定的順序；
SynchronousQueue:特殊的BlockingQueue,對其的操作必須是放和取交替完成的；
其中LinkedBlockingQueue和ArrayBlockingQueue比較起來,它們背后所用的數據結構不一樣,導致LinkedBlockingQueue的數據吞吐量要大於ArrayBlockingQueue,
但在線程數量很大時其性能的可預見性低於ArrayBlockingQueue。
四、BlockingQueue的幾個注意點
1.BlockingQueue可以是限定容量的。
它在任意給定時間都可以有一個remainingCapacity，超出此容量，便無法無阻塞地put 附加元素。沒有任何內部容量約束的BlockingQueue 總是報告Integer.MAX_VALUE 的剩余容量。
2.BlockingQueue實現主要用於生產者-使用者隊列，還支持Collection接口。
舉例來說，使用remove(x)從隊列中移除任意一個元素是有可能的。然而，這種操作通常不會有效執行，只能有計划地偶爾使用，比如在取消排隊信息時。
3.BlockingQueue實現是線程安全的。
所有排隊方法都可以使用內部鎖或其他形式的並發控制來自動達到它們的目的。然而，大量的Collection 操作（addAll、containsAll、retainAll 和removeAll）沒有必要自動執行，除非在實現中特別說明。因此，舉例來說，在只添加了c中的一些元素后，addAll(c)有可能失敗（拋出一個異常）。
4.BlockingQueue實質上不支持使用任何一種“close”或“shutdown”操作來指示不再添加任何項。
這種功能的需求和使用有依賴於實現的傾向。例如，一種常用的策略是：對於生產者，插入特殊的end-of-stream 或poison 對象，並根據使用者獲取這些對象的時間來對它們進行解釋。
參考文章：http://wsmajunfeng.iteye.com/blog/1629354
http://zzhonghe.iteye.com/blog/826757
http://blog.csdn.net/xin_jmail/article/details/26157971