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

什么是阻塞隊(duì)列？

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

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

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

（2）支持阻塞的移除方法：意思是在隊(duì)列為空時(shí)，獲取元素的線程會(huì)等待隊(duì)列變?yōu)榉强?。阻塞?duì)列常用于生產(chǎn)者和消費(fèi)者的場(chǎng)景，生產(chǎn)者是向隊(duì)列里添加元素的線程，消費(fèi)者是從隊(duì)列里取元素的線程。阻塞隊(duì)列就是生產(chǎn)者用來(lái)存放元素、消費(fèi)者用來(lái)獲取元素的容器。

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

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

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

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

實(shí)現(xiàn)阻塞接口的隊(duì)列：

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

五個(gè)隊(duì)列所提供的各有不同：

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

·LinkedBlockingQueue ：一個(gè)由鏈接節(jié)點(diǎn)支持的可選有界隊(duì)列。

·PriorityBlockingQueue ：一個(gè)由優(yōu)先級(jí)堆支持的無(wú)界優(yōu)先級(jí)隊(duì)列。

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

·SynchronousQueue ：一個(gè)利用 BlockingQueue 接口的簡(jiǎn)單聚集（rendezvous）機(jī)制。

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

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

簡(jiǎn)單介紹下其中的幾個(gè)：

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

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

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

關(guān)于ConcurrentLinkedQueue和LinkedBlockingQueue：

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

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

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

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

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

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

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() + "個(gè)");
                        Thread.sleep(300);
                    }
                } catch (InterruptedException e) {}
            }
        }
 
        class Consumer implements Runnable {
            public void run() {
                try {
                    while (true) {
                        System.out.println("消費(fèi)者開始消費(fèi)蘋果***");
                        basket.consume();
                        System.out.println("消費(fèi)者消費(fèi)蘋果完畢***");
                        System.out.println("籃子中的蘋果數(shù)量：" + basket.getAppleNumber() + "個(gè)");
                        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í)！

以上是java 隊(duì)列是什么的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！