隊列是一種特殊的線性表，遵循的原則就是“先入先出”。在我們?nèi)粘Ｊ褂弥?，?jīng)常會用來并發(fā)操作數(shù)據(jù)。在并發(fā)編程中，有時候需要使用線程安全的隊列。如果要實現(xiàn)一個線程安全的隊列通常有兩種方式：一種是使用阻塞隊列，另一種是使用線程同步鎖。

什么是阻塞隊列？

假設(shè)有一個面包房，里面有一個客人吃面包，一個師傅烤面包。籃子里面最多放2個面包，師傅考完了面包放到籃子里，而客人吃面包則從籃子里面往外拿，為了保證客人吃面包的時候籃子里有面包或者師傅烤面包的時候籃子不會溢出，這時候就需要引用出來阻塞隊列的概念，就是我們常說的生產(chǎn)者消費者的模式。

阻塞隊列是一個支持兩個附加操作的隊列。這兩個附加的操作支持阻塞的插入和移除方法。

（1）支持阻塞的插入方法：意思是當(dāng)隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。

（2）支持阻塞的移除方法：意思是在隊列為空時，獲取元素的線程會等待隊列變?yōu)榉强铡Ｗ枞犃谐Ｓ糜谏a(chǎn)者和消費者的場景，生產(chǎn)者是向隊列里添加元素的線程，消費者是從隊列里取元素的線程。阻塞隊列就是生產(chǎn)者用來存放元素、消費者用來獲取元素的容器。

系統(tǒng)內(nèi)不阻塞隊列：PriorityQueue 和 ConcurrentLinkedQueue

我們來看一下不阻塞隊列的關(guān)系（以PriorityQueue 為例）：

PriorityQueue 類繼承自AbstractQueue，實現(xiàn)了Serializable接口。實質(zhì)上維護了一個有序列表，PriorityQueue位于Java util包中，觀其名字前半部分的單詞Priority是優(yōu)先的意思，實際上這個隊列就是具有“優(yōu)先級”。加入到 Queue 中的元素根據(jù)它們的天然排序（通過其 java.util.Comparable 實現(xiàn)）或者根據(jù)傳遞給構(gòu)造函數(shù)的 java.util.Comparator 實現(xiàn)來定位。

ConcurrentLinkedQueue 是基于鏈接節(jié)點的、線程安全的隊列。并發(fā)訪問不需要同步。因為它在隊列的尾部添加元素并從頭部刪除它們，所以不需要知道隊列的大小， ConcurrentLinkedQueue 對公共集合的共享訪問就可以工作得很好。收集關(guān)于隊列大小的信息會很慢，需要遍歷隊列；ConcurrentLinkedQueue是一個基于鏈接節(jié)點的無界線程安全隊列，它采用先進先出的規(guī)則對節(jié)點進行排序，當(dāng)我們添加一個元素的時候，它會添加到隊列的尾部；當(dāng)我們獲取一個元素時，它會返回隊列頭部的元素。

實現(xiàn)阻塞接口的隊列：

java.util.concurrent 中加入了 BlockingQueue 接口和五個阻塞隊列類。它實質(zhì)上就是一種帶有一點扭曲的 FIFO 數(shù)據(jù)結(jié)構(gòu)。不是立即從隊列中添加或者刪除元素，線程執(zhí)行操作阻塞，直到有空間或者元素可用。

五個隊列所提供的各有不同：

·ArrayBlockingQueue ：一個由數(shù)組支持的有界隊列。

·LinkedBlockingQueue ：一個由鏈接節(jié)點支持的可選有界隊列。

·PriorityBlockingQueue ：一個由優(yōu)先級堆支持的無界優(yōu)先級隊列。

·DelayQueue ：一個由優(yōu)先級堆支持的、基于時間的調(diào)度隊列。

·SynchronousQueue ：一個利用 BlockingQueue 接口的簡單聚集（rendezvous）機制。

我們看一下ArrayBlockingQueue 和LinkedBlockingQueue 的繼承關(guān)系：

通過查看兩個類的繼承關(guān)系，我們可以知道，他們也是繼承自AbstractQueue，實現(xiàn)了Serializable接口；不同的是他們同時實現(xiàn)了BlockingQueue接口。

簡單介紹下其中的幾個：

LinkedBlockingQueueLinkedBlockingQueue默認(rèn)大小是Integer.MAX_VALUE，可以理解為一個緩存的有界等待隊列，可以選擇指定其最大容量，它是基于鏈表的隊列，此隊列按 FIFO（先進先出）排序元素。當(dāng)生產(chǎn)者往隊列中放入一個數(shù)據(jù)時，緩存在隊列內(nèi)部，當(dāng)隊列緩沖區(qū)達到最大值緩存容量時（LinkedBlockingQueue可以通過構(gòu)造函數(shù)指定該值），阻塞生產(chǎn)者隊列，直到消費者從隊列中消費掉一份數(shù)據(jù)，生產(chǎn)者線程會被喚醒，反之對于消費者同理。

ArrayBlockingQueue在構(gòu)造時需要指定容量， 并可以選擇是否需要公平性，如果公平參數(shù)被設(shè)置true，等待時間最長的線程會優(yōu)先得到處理（其實就是通過將ReentrantLock設(shè)置為true來 達到這種公平性的：即等待時間最長的線程會先操作）。通常，公平性會使你在性能上付出代價，只有在的確非常需要的時候再使用它。它是基于數(shù)組的阻塞循環(huán)隊列，此隊列按FIFO（先進先出）原則對元素進行排序。

PriorityBlockingQueue是一個帶優(yōu)先級的 隊列，而不是先進先出隊列。元素按優(yōu)先級順序被移除，該隊列也沒有上限（看了一下源碼，PriorityBlockingQueue是對 PriorityQueue的再次包裝，是基于堆數(shù)據(jù)結(jié)構(gòu)的，而PriorityQueue是沒有容量限制的，與ArrayList一樣，所以在優(yōu)先阻塞 隊列上put時是不會受阻的。雖然此隊列邏輯上是無界的，但是由于資源被耗盡，所以試圖執(zhí)行添加操作可能會導(dǎo)致 OutOfMemoryError），但是如果隊列為空，那么取元素的操作take就會阻塞，所以它的檢索操作take是受阻的。另外，往入該隊列中的元 素要具有比較能力。

關(guān)于ConcurrentLinkedQueue和LinkedBlockingQueue：

也可以理解為阻塞隊列和非阻塞隊列的區(qū)別：

1.LinkedBlockingQueue是使用鎖機制，ConcurrentLinkedQueue是使用CAS算法，雖然LinkedBlockingQueue的底層獲取鎖也是使用的CAS算法。

2.關(guān)于取元素，ConcurrentLinkedQueue不支持阻塞去取元素，LinkedBlockingQueue支持阻塞的take()方法。

3.關(guān)于插入元素的性能，但在實際的使用過程中，尤其在多cpu的服務(wù)器上，有鎖和無鎖的差距便體現(xiàn)出來了，ConcurrentLinkedQueue會比LinkedBlockingQueue快很多。

生產(chǎn)者消費者代碼：

在網(wǎng)上看到一個生產(chǎn)者消費者的小例子，對于理解阻塞隊列非常有幫助，代碼如下：

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;   public class BlockingQueueTest {     public static class Basket {         BlockingQueue<String> basket = new ArrayBlockingQueue<>(3);           private void produce() throws InterruptedException {             basket.put("蘋果");         }           private void consume() throws InterruptedException {             basket.take();         }           private int getAppleNumber() {             return basket.size();         }     }       private static void testBasket() {         final Basket basket = new Basket();         class Producer implements Runnable {             public void run() {                 try {                     while (true) {                         System.out.println("生產(chǎn)者開始生產(chǎn)蘋果###");                         basket.produce();                         System.out.println("生產(chǎn)者生產(chǎn)蘋果完畢###");                         System.out.println("籃子中的蘋果數(shù)量：" + basket.getAppleNumber() + "個");                         Thread.sleep(300);                     }                 } catch (InterruptedException e) {}             }         }           class Consumer implements Runnable {             public void run() {                 try {                     while (true) {                         System.out.println("消費者開始消費蘋果***");                         basket.consume();                         System.out.println("消費者消費蘋果完畢***");                         System.out.println("籃子中的蘋果數(shù)量：" + basket.getAppleNumber() + "個");                         Thread.sleep(1000);                     }                 } catch (InterruptedException e) {}             }         }         ExecutorService service = Executors.newCachedThreadPool();         Producer producer = new Producer();         Consumer consumer = new Consumer();         service.submit(producer);         service.submit(consumer);         try {             Thread.sleep(10000);         } catch (InterruptedException e) {}         service.shutdownNow();     }       public static void main(String[] args) {         BlockingQueueTest.testBasket();     } }

眾多java培訓(xùn)視頻，盡在PHP中文網(wǎng)，歡迎在線學(xué)習(xí)！