前言

本節(jié)就一起探討一下什么是阻塞隊(duì)列，一起來看下吧！

什么是阻塞隊(duì)列

我們先說一下為什么要引入阻塞隊(duì)列。我們知道服務(wù)器的資源是有限的，就拿典型的生產(chǎn)者和消費(fèi)者模型來講。

消費(fèi)者如果沒有東西可以消費(fèi)了，但是它還是在一直執(zhí)行，這無疑是在浪費(fèi)系統(tǒng)資源，所以我們需要阻塞消費(fèi)者，換過來同理。

生產(chǎn)者沒有東西可生產(chǎn)了或者說沒有地方存儲(chǔ)它生產(chǎn)的東西，這時(shí)候我們就需要阻塞生產(chǎn)者。

然而，在開發(fā)中，這種模型往往運(yùn)行在多線程環(huán)境中，需要對資源共享從而達(dá)到更高的性能，但這樣也會(huì)造成線程安全問題，比如死鎖、重復(fù)消費(fèi) 所以阻塞隊(duì)列就是幫我們解決這些問題的。

BlockingQueue

下面我們就一起看一下Java中給我們提供了哪些好用的工具，先給大家介紹一下BlockingQueue, 本節(jié)我們主要講它的用法。

BlockingQueue是Java util.concurrent包下的類，BlockingQueue提供了「線程安全的隊(duì)列訪問方式」，并發(fā)包下很多高級(jí)同步類的實(shí)現(xiàn)都是基于BlockingQueue實(shí)現(xiàn)的。

BlockingQueue一般用于生產(chǎn)者-消費(fèi)者模式，生產(chǎn)者是往隊(duì)列里添加元素的線程，消費(fèi)者是從隊(duì)列里拿元素的線程。「BlockingQueue就是存放元素的容器」。

public interface BlockingQueue<E> extends Queue<E> {....}

它本身是一個(gè)接口，我們看下它常用的實(shí)現(xiàn)類。

ArrayBlockingQueue

一個(gè)用數(shù)組實(shí)現(xiàn)的有界阻塞隊(duì)列，此隊(duì)列按照先進(jìn)先出（FIFO）的原則對元素進(jìn)行排序, 支持公平鎖和非公平鎖。

• add() 添加元素。

// 初始化 容量為3
BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
System.out.println(queue.add("1"));
System.out.println(queue.add("2"));
System.out.println(queue.add("3"));
System.out.println(queue.add("4"));

我們發(fā)現(xiàn)添加到4的時(shí)候，報(bào)錯(cuò)了Exception in thread "main" java.lang.IllegalStateException: Queue full。

• remove() 移除元素,如果為空，會(huì)返回異常。

// 初始化 容量為3
BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
System.out.println(queue.add("1"));
System.out.println(queue.add("2"));
System.out.println(queue.add("3"));
// System.out.println(queue.add("4"));
System.out.println(queue.remove());
System.out.println(queue);

打印:

true
true
true
1
[2, 3]

從結(jié)果來看，符合FIFO規(guī)則,如果我想移除指定元素怎么辦呢?很簡單,使用remove("3")。

• offer() 和add方法類似，但是它不會(huì)報(bào)錯(cuò)，它會(huì)返回false， offer(e,time,unit)支持超時(shí)。

BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
System.out.println(queue.offer("1"));
System.out.println(queue.offer("2"));
System.out.println(queue.offer("3"));
System.out.println(queue.offer("4"));

true
true
true
false

• poll() 與remove相似，但它不會(huì)返回異常而是返回null, poll(time,unit)。

BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
System.out.println(queue.poll());

null

• take() 與 poll類似, 但它會(huì)造成線程阻塞。

public static void main(String[] args) throws InterruptedException {
    BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
    System.out.println(queue.take());
}

運(yùn)行后，我們發(fā)現(xiàn)程序被阻塞了。

• put() 添加元素，當(dāng)容器滿的時(shí)候,會(huì)造成線程阻塞。

public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
        queue.put("1");
        queue.put("2");
    }

運(yùn)行后發(fā)現(xiàn)，執(zhí)行到2的時(shí)候被阻塞了。

以上是它的常用方法，其它的實(shí)現(xiàn)類就不一一給大家演示了，方法差不多，都是基于BlockingQueue接口,可以自己試著運(yùn)行一下看看。

LinkedBlockingQueue

一個(gè)由鏈表結(jié)構(gòu)組成的有界隊(duì)列，此隊(duì)列的長度為Integer.MAX_VALUE。

SynchronousQueue

是一個(gè)不存儲(chǔ)元素的阻塞隊(duì)列，每一個(gè)put操作必須等待take操作，否則不能添加元素。支持公平鎖和非公平鎖。SynchronousQueue的一個(gè)使用場景是在線程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue，這個(gè)上節(jié)我們講線程復(fù)用的時(shí)候遇到過。

LinkedTransferQueue

是一個(gè)由鏈表結(jié)構(gòu)組成的無界阻塞隊(duì)列，相當(dāng)于其它隊(duì)列，LinkedTransferQueue隊(duì)列多了transfer和tryTransfer方法。

PriorityBlockingQueue

是一個(gè)支持線程優(yōu)先級(jí)排序的無界隊(duì)列，默認(rèn)自然序進(jìn)行排序，也可以自定義實(shí)現(xiàn)compareTo()方法來指定元素排序規(guī)則，不能保證同優(yōu)先級(jí)元素的順序。

DelayQueue

一個(gè)實(shí)現(xiàn)PriorityBlockingQueue實(shí)現(xiàn)延遲獲取的無界隊(duì)列，在創(chuàng)建元素時(shí)，可以指定多久才能從隊(duì)列中獲取當(dāng)前元素。只有延時(shí)期滿后才能從隊(duì)列中獲取元素，在緩存設(shè)計(jì)和定時(shí)任務(wù)調(diào)度中經(jīng)常會(huì)遇到。

結(jié)束語

本節(jié)主要給大家介紹了常用的阻塞隊(duì)列以及它的基本使用。