這是java高併發系列第25篇文章。

環境：jdk1.8。

本文內容

1. 掌握Queue、BlockingQueue介面中常用的方法
2. 介紹6中阻塞佇列，及相關場景示例
3. 重點掌握4種常用的阻塞佇列

Queue介面

佇列是一種先進先出（FIFO）的資料結構，java中用Queue介面來表示佇列。

Queue介面中定義了6個方法：

public interface Queue<E> extends Collection<E> {
    boolean add(e);
    boolean offer(E e);
    E remove();
    E poll();
    E element();
    E peek();
}
 

每個Queue方法都有兩種形式：

（1）如果操作失敗則丟擲異常，

（2）如果操作失敗，則返回特殊值（null或false，具體取決於操作），介面的常規結構如下表所示。

操作型別	丟擲異常	返回特殊值
插入	add(e)	offer(e)
移除	remove()	poll()
檢查	element()	peek()

Queue從Collection繼承的add方法插入一個元素，除非它違反了佇列的容量限制，在這種情況下它會丟擲IllegalStateException；offer方法與add不同之處僅在於它通過返回false來表示插入元素失敗。

remove和poll

方法都移除並返回佇列的頭部，確切地移除哪個元素是由具體的實現來決定的，僅當佇列為空時，remove和poll方法的行為才有所不同，在這些情況下，remove丟擲NoSuchElementException，而poll返回null。

element和peek方法返回佇列頭部的元素，但不移除，它們之間的差異與remove和poll的方式完全相同，如果佇列為空，則element丟擲NoSuchElementException，而peek返回null。

佇列一般不要插入空元素。

BlockingQueue介面

BlockingQueue位於juc中，熟稱阻塞佇列， 阻塞佇列首先它是一個佇列，繼承Queue

介面，是佇列就會遵循先進先出（FIFO）的原則，又因為它是阻塞的，故與普通的佇列有兩點區別：

1. 當一個執行緒向佇列裡面新增資料時，如果佇列是滿的，那麼將阻塞該執行緒，暫停新增資料
2. 當一個執行緒從佇列裡面取出資料時，如果佇列是空的，那麼將阻塞該執行緒，暫停取出資料

BlockingQueue相關方法：

操作型別	丟擲異常	返回特殊值	一直阻塞	超時退出
插入	add(e)	offer(e)	put(e)	offer(e,timeuout,unit)
移除	remove()	poll()	take()	poll(timeout,unit)
檢查	element()	peek()	不支援	不支援

重點，再來解釋一下，加深印象：

1. 3個可能會有異常的方法，add、remove、element；這3個方法不會阻塞（是說佇列滿或者空的情況下是否會阻塞）；佇列滿的情況下，add丟擲異常；佇列為空情況下，remove、element丟擲異常
2. offer、poll、peek 也不會阻塞（是說佇列滿或者空的情況下是否會阻塞）；佇列滿的情況下，offer返回false；佇列為空的情況下，pool、peek返回null
3. 佇列滿的情況下，呼叫put方法會導致當前執行緒阻塞
4. 佇列為空的情況下，呼叫take方法會導致當前執行緒阻塞
5. offer(e,timeuout,unit)，超時之前，插入成功返回true，否者返回false
6. poll(timeout,unit)，超時之前，獲取到頭部元素並將其移除，返回true，否者返回false
7. 以上一些方法希望大家都記住，方便以後使用

BlockingQueue常見的實現類

看一下相關類圖

ArrayBlockingQueue

基於陣列的阻塞佇列實現，其內部維護一個定長的陣列，用於儲存佇列元素。執行緒阻塞的實現是通過ReentrantLock來完成的，資料的插入與取出共用同一個鎖，因此ArrayBlockingQueue並不能實現生產、消費同時進行。而且在建立ArrayBlockingQueue時，我們還可以控制物件的內部鎖是否採用公平鎖，預設採用非公平鎖。

LinkedBlockingQueue

基於單向連結串列的阻塞佇列實現，在初始化LinkedBlockingQueue的時候可以指定大小，也可以不指定，預設類似一個無限大小的容量（Integer.MAX_VALUE），不指佇列容量大小也是會有風險的，一旦資料生產速度大於消費速度，系統記憶體將有可能被消耗殆盡，因此要謹慎操作。另外LinkedBlockingQueue中用於阻塞生產者、消費者的鎖是兩個（鎖分離），因此生產與消費是可以同時進行的。

PriorityBlockingQueue

一個支援優先順序排序的無界阻塞佇列，進入佇列的元素會按照優先順序進行排序

SynchronousQueue

同步阻塞佇列，SynchronousQueue沒有容量，與其他BlockingQueue不同，SynchronousQueue是一個不儲存元素的BlockingQueue，每一個put操作必須要等待一個take操作，否則不能繼續新增元素，反之亦然

DelayQueue

DelayQueue是一個支援延時獲取元素的無界阻塞佇列，裡面的元素全部都是“可延期”的元素，列頭的元素是最先“到期”的元素，如果佇列裡面沒有元素到期，是不能從列頭獲取元素的，哪怕有元素也不行，也就是說只有在延遲期到時才能夠從佇列中取元素

LinkedTransferQueue

LinkedTransferQueue是基於連結串列的FIFO無界阻塞佇列，它出現在JDK7中，Doug Lea 大神說LinkedTransferQueue是一個聰明的佇列，它是ConcurrentLinkedQueue、SynchronousQueue(公平模式下)、無界的LinkedBlockingQueues等的超集，LinkedTransferQueue包含了ConcurrentLinkedQueue、SynchronousQueue、LinkedBlockingQueues三種佇列的功能

下面我們來介紹每種阻塞佇列的使用。

ArrayBlockingQueue

有界阻塞佇列，內部使用陣列儲存元素，有2個常用構造方法：

//capacity表示容量大小，預設內部採用非公平鎖
public ArrayBlockingQueue(int capacity)
//capacity：容量大小，fair：內部是否是使用公平鎖
public ArrayBlockingQueue(int capacity, boolean fair)

需求：業務系統中有很多地方需要推送通知，由於需要推送的資料太多，我們將需要推送的資訊先丟到阻塞佇列中，然後開一個執行緒進行處理真實發送，程式碼如下：

package com.itsoku.chat25;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import sun.text.normalizer.NormalizerBase;

import java.util.Calendar;
import java.util.concurrent.*;

/**
 * 跟著阿里p7學併發，微信公眾號：javacode2018
 */
public class Demo1 {
    //推送佇列
    static ArrayBlockingQueue<String> pushQueue = new ArrayBlockingQueue<String>(10000);

    static {
        //啟動一個執行緒做真實推送
        new Thread(() -> {
            while (true) {
                String msg;
                try {
                    long starTime = System.currentTimeMillis();
                    //獲取一條推送訊息，此方法會進行阻塞，直到返回結果
                    msg = pushQueue.take();
                    long endTime = System.currentTimeMillis();
                    //模擬推送耗時
                    TimeUnit.MILLISECONDS.sleep(500);

                    System.out.println(String.format("[%s,%s,take耗時:%s],%s,傳送訊息:%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName(), msg));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    //推送訊息，需要傳送推送訊息的呼叫該方法，會將推送資訊先加入推送佇列
    public static void pushMsg(String msg) throws InterruptedException {
        pushQueue.put(msg);
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 1; i <= 5; i++) {
            String msg = "一起來學java高併發,第" + i + "天";
            //模擬耗時
            TimeUnit.SECONDS.sleep(i);
            Demo1.pushMsg(msg);
        }
    }
}

輸出：

[1565595629206,1565595630207,take耗時:1001],Thread-0,傳送訊息:一起來學java高併發,第1天
[1565595630208,1565595632208,take耗時:2000],Thread-0,傳送訊息:一起來學java高併發,第2天
[1565595632208,1565595635208,take耗時:3000],Thread-0,傳送訊息:一起來學java高併發,第3天
[1565595635208,1565595639209,take耗時:4001],Thread-0,傳送訊息:一起來學java高併發,第4天
[1565595639209,1565595644209,take耗時:5000],Thread-0,傳送訊息:一起來學java高併發,第5天

程式碼中我們使用了有界佇列ArrayBlockingQueue，建立ArrayBlockingQueue時候需要制定容量大小，呼叫pushQueue.put將推送資訊放入佇列中，如果佇列已滿，此方法會阻塞。程式碼中在靜態塊中啟動了一個執行緒，呼叫pushQueue.take();從佇列中獲取待推送的資訊進行推送處理。

注意：ArrayBlockingQueue如果佇列容量設定的太小，消費者傳送的太快，消費者消費的太慢的情況下，會導致佇列空間滿，呼叫put方法會導致傳送者執行緒阻塞，所以注意設定合理的大小，協調好消費者的速度。

LinkedBlockingQueue

內部使用單向連結串列實現的阻塞佇列，3個構造方法：

//預設構造方法，容量大小為Integer.MAX_VALUE
public LinkedBlockingQueue();
//建立指定容量大小的LinkedBlockingQueue
public LinkedBlockingQueue(int capacity);
//容量為Integer.MAX_VALUE,並將傳入的集合丟入佇列中
public LinkedBlockingQueue(Collection<? extends E> c);

LinkedBlockingQueue的用法和ArrayBlockingQueue類似，建議使用的時候指定容量，如果不指定容量，插入的太快，移除的太慢，可能會產生OOM。

PriorityBlockingQueue

無界的優先順序阻塞佇列，內部使用陣列儲存資料，達到容量時，會自動進行擴容，放入的元素會按照優先順序進行排序，4個構造方法：

//預設構造方法，預設初始化容量是11
public PriorityBlockingQueue();
//指定佇列的初始化容量
public PriorityBlockingQueue(int initialCapacity);
//指定佇列的初始化容量和放入元素的比較器
public PriorityBlockingQueue(int initialCapacity,Comparator<? super E> comparator);
//傳入集合放入來初始化佇列，傳入的集合可以實現SortedSet介面或者PriorityQueue介面進行排序，如果沒有實現這2個介面，按正常順序放入佇列
public PriorityBlockingQueue(Collection<? extends E> c);

優先順序佇列放入元素的時候，會進行排序，所以我們需要指定排序規則，有2種方式：

1. 建立PriorityBlockingQueue指定比較器Comparator
2. 放入的元素需要實現Comparable介面

上面2種方式必須選一個，如果2個都有，則走第一個規則排序。

需求：還是上面的推送業務，目前推送是按照放入的先後順序進行傳送的，比如有些公告比較緊急，優先順序比較高，需要快點發送，怎麼搞？此時PriorityBlockingQueue就派上用場了，程式碼如下：

package com.itsoku.chat25;

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 跟著阿里p7學併發，微信公眾號：javacode2018
 */
public class Demo2 {

    //推送資訊封裝
    static class Msg implements Comparable<Msg> {
        //優先順序，越小優先順序越高
        private int priority;
        //推送的資訊
        private String msg;

        public Msg(int priority, String msg) {
            this.priority = priority;
            this.msg = msg;
        }

        @Override
        public int compareTo(Msg o) {
            return Integer.compare(this.priority, o.priority);
        }

        @Override
        public String toString() {
            return "Msg{" +
                    "priority=" + priority +
                    ", msg='" + msg + '\'' +
                    '}';
        }
    }

    //推送佇列
    static PriorityBlockingQueue<Msg> pushQueue = new PriorityBlockingQueue<Msg>();

    static {
        //啟動一個執行緒做真實推送
        new Thread(() -> {
            while (true) {
                Msg msg;
                try {
                    long starTime = System.currentTimeMillis();
                    //獲取一條推送訊息，此方法會進行阻塞，直到返回結果
                    msg = pushQueue.take();
                    //模擬推送耗時
                    TimeUnit.MILLISECONDS.sleep(100);
                    long endTime = System.currentTimeMillis();
                    System.out.println(String.format("[%s,%s,take耗時:%s],%s,傳送訊息:%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName(), msg));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    //推送訊息，需要傳送推送訊息的呼叫該方法，會將推送資訊先加入推送佇列
    public static void pushMsg(int priority, String msg) throws InterruptedException {
        pushQueue.put(new Msg(priority, msg));
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 5; i >= 1; i--) {
            String msg = "一起來學java高併發,第" + i + "天";
            Demo2.pushMsg(i, msg);
        }
    }
}

輸出：

[1565598857028,1565598857129,take耗時:101],Thread-0,傳送訊息:Msg{priority=1, msg='一起來學java高併發,第1天'}
[1565598857162,1565598857263,take耗時:101],Thread-0,傳送訊息:Msg{priority=2, msg='一起來學java高併發,第2天'}
[1565598857263,1565598857363,take耗時:100],Thread-0,傳送訊息:Msg{priority=3, msg='一起來學java高併發,第3天'}
[1565598857363,1565598857463,take耗時:100],Thread-0,傳送訊息:Msg{priority=4, msg='一起來學java高併發,第4天'}
[1565598857463,1565598857563,take耗時:100],Thread-0,傳送訊息:Msg{priority=5, msg='一起來學java高併發,第5天'}

main中放入了5條推送資訊，i作為訊息的優先順序按倒敘放入的，最終輸出結果中按照優先順序由小到大輸出。注意Msg實現了Comparable介面，具有了比較功能。

SynchronousQueue

同步阻塞佇列，SynchronousQueue沒有容量，與其他BlockingQueue不同，SynchronousQueue是一個不儲存元素的BlockingQueue，每一個put操作必須要等待一個take操作，否則不能繼續新增元素，反之亦然。SynchronousQueue 在現實中用的不多，執行緒池中有用到過，Executors.newCachedThreadPool()實現中用到了這個佇列，當有任務丟入執行緒池的時候，如果已建立的工作執行緒都在忙於處理任務，則會新建一個執行緒來處理丟入佇列的任務。

來個示例程式碼：

package com.itsoku.chat25;

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * 跟著阿里p7學併發，微信公眾號：javacode2018
 */
public class Demo3 {

    static SynchronousQueue<String> queue = new SynchronousQueue<>();

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            try {
                long starTime = System.currentTimeMillis();
                queue.put("java高併發系列，路人甲Java!");
                long endTime = System.currentTimeMillis();
                System.out.println(String.format("[%s,%s,take耗時:%s],%s", starTime, endTime, (endTime - starTime), Thread.currentThread().getName()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        //休眠5秒之後，從佇列中take一個元素
        TimeUnit.SECONDS.sleep(5);
        System.out.println(System.currentTimeMillis() + "呼叫take獲取並移除元素," + queue.take());
    }
}

輸出：

1565600421645呼叫take獲取並移除元素,java高併發系列，路人甲Java!
[1565600416645,1565600421645,take耗時:5000],Thread-0

main方法中啟動了一個執行緒，呼叫queue.put方法向佇列中丟入一條資料，呼叫的時候產生了阻塞，從輸出結果中可以看出，直到take方法被呼叫時，put方法才從阻塞狀態恢復正常。

DelayQueue

DelayQueue是一個支援延時獲取元素的無界阻塞佇列，裡面的元素全部都是“可延期”的元素，列頭的元素是最先“到期”的元素，如果佇列裡面沒有元素到期，是不能從列頭獲取元素的，哪怕有元素也不行，也就是說只有在延遲期到時才能夠從佇列中取元素。

需求：還是推送的業務，有時候我們希望早上9點或者其他指定的時間進行推送，如何實現呢？此時DelayQueue就派上用場了。

我們先看一下DelayQueue類的宣告：

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E>

元素E需要實現介面Delayed，我們看一下這個介面的程式碼：

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

Delayed繼承了Comparable介面，這個介面是用來做比較用的，DelayQueue內部使用PriorityQueue來儲存資料的，PriorityQueue是一個優先順序佇列，丟入的資料會進行排序，排序方法呼叫的是Comparable介面中的方法。下面主要說一下Delayed介面中的getDelay方法：此方法在給定的時間單位內返回與此物件關聯的剩餘延遲時間。

對推送我們再做一下處理，讓其支援定時傳送（定時在將來某個時間也可以說是延遲傳送），程式碼如下：

package com.itsoku.chat25;

import java.util.Calendar;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 跟著阿里p7學併發，微信公眾號：javacode2018
 */
public class Demo4 {

    //推送資訊封裝
    static class Msg implements Delayed {
        //優先順序，越小優先順序越高
        private int priority;
        //推送的資訊
        private String msg;
        //定時傳送時間，毫秒格式
        private long sendTimeMs;

        public Msg(int priority, String msg, long sendTimeMs) {
            this.priority = priority;
            this.msg = msg;
            this.sendTimeMs = sendTimeMs;
        }

        @Override
        public String toString() {
            return "Msg{" +
                    "priority=" + priority +
                    ", msg='" + msg + '\'' +
                    ", sendTimeMs=" + sendTimeMs +
                    '}';
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.sendTimeMs - Calendar.getInstance().getTimeInMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            if (o instanceof Msg) {
                Msg c2 = (Msg) o;
                return Integer.compare(this.priority, c2.priority);
            }
            return 0;
        }
    }

    //推送佇列
    static DelayQueue<Msg> pushQueue = new DelayQueue<Msg>();

    static {
        //啟動一個執行緒做真實推送
        new Thread(() -> {
            while (true) {
                Msg msg;
                try {
                    //獲取一條推送訊息，此方法會進行阻塞，直到返回結果
                    msg = pushQueue.take();
                    //此處可以做真實推送
                    long endTime = System.currentTimeMillis();
                    System.out.println(String.format("定時傳送時間：%s,實際傳送時間：%s,傳送訊息:%s", msg.sendTimeMs, endTime, msg));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    //推送訊息，需要傳送推送訊息的呼叫該方法，會將推送資訊先加入推送佇列
    public static void pushMsg(int priority, String msg, long sendTimeMs) throws InterruptedException {
        pushQueue.put(new Msg(priority, msg, sendTimeMs));
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 5; i >= 1; i--) {
            String msg = "一起來學java高併發,第" + i + "天";
            Demo4.pushMsg(i, msg, Calendar.getInstance().getTimeInMillis() + i * 2000);
        }
    }
}

輸出：

定時傳送時間：1565603357198,實際傳送時間：1565603357198,傳送訊息:Msg{priority=1, msg='一起來學java高併發,第1天', sendTimeMs=1565603357198}
定時傳送時間：1565603359198,實際傳送時間：1565603359198,傳送訊息:Msg{priority=2, msg='一起來學java高併發,第2天', sendTimeMs=1565603359198}
定時傳送時間：1565603361198,實際傳送時間：1565603361199,傳送訊息:Msg{priority=3, msg='一起來學java高併發,第3天', sendTimeMs=1565603361198}
定時傳送時間：1565603363198,實際傳送時間：1565603363199,傳送訊息:Msg{priority=4, msg='一起來學java高併發,第4天', sendTimeMs=1565603363198}
定時傳送時間：1565603365182,實際傳送時間：1565603365183,傳送訊息:Msg{priority=5, msg='一起來學java高併發,第5天', sendTimeMs=1565603365182}

可以看出時間傳送時間，和定時傳送時間基本一致，程式碼中Msg需要實現Delayed介面，重點在於getDelay方法，這個方法返回剩餘的延遲時間，程式碼中使用this.sendTimeMs減去當前時間的毫秒格式時間，得到剩餘延遲時間。

LinkedTransferQueue

LinkedTransferQueue是一個由連結串列結構組成的無界阻塞TransferQueue佇列。相對於其他阻塞佇列，LinkedTransferQueue多了tryTransfer和transfer方法。

LinkedTransferQueue類繼承自AbstractQueue抽象類，並且實現了TransferQueue介面：

public interface TransferQueue<E> extends BlockingQueue<E> {
    // 如果存在一個消費者已經等待接收它，則立即傳送指定的元素，否則返回false，並且不進入佇列。
    boolean tryTransfer(E e);
    // 如果存在一個消費者已經等待接收它，則立即傳送指定的元素，否則等待直到元素被消費者接收。
    void transfer(E e) throws InterruptedException;
    // 在上述方法的基礎上設定超時時間
    boolean tryTransfer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;
    // 如果至少有一位消費者在等待，則返回true
    boolean hasWaitingConsumer();
    // 獲取所有等待獲取元素的消費執行緒數量
    int getWaitingConsumerCount();
}

再看一下上面的這些方法，transfer(E e)方法和SynchronousQueue的put方法類似，都需要等待消費者取走元素，否者一直等待。其他方法和ArrayBlockingQueue、LinkedBlockingQueue中的方法類似。

總結

1. 重點需要了解BlockingQueue中的所有方法，以及他們的區別
2. 重點掌握ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue的使用場景
3. 需要處理的任務有優先順序的，使用PriorityBlockingQueue
4. 處理的任務需要延時處理的，使用DelayQueue

java高併發系列目錄

1. 第1天:必須知道的幾個概念
2. 第2天:併發級別
3. 第3天:有關並行的兩個重要定律
4. 第4天:JMM相關的一些概念
5. 第5天:深入理解程序和執行緒
6. 第6天:執行緒的基本操作
7. 第7天:volatile與Java記憶體模型
8. 第8天:執行緒組
9. 第9天：使用者執行緒和守護執行緒
10. 第10天:執行緒安全和synchronized關鍵字
11. 第11天:執行緒中斷的幾種方式
12. 第12天JUC:ReentrantLock重入鎖
13. 第13天:JUC中的Condition物件
14. 第14天:JUC中的LockSupport工具類，必備技能
15. 第15天：JUC中的Semaphore（訊號量）
16. 第16天：JUC中等待多執行緒完成的工具類CountDownLatch，必備技能
17. 第17天：JUC中的迴圈柵欄CyclicBarrier的6種使用場景
18. 第18天：JAVA執行緒池，這一篇就夠了
19. 第19天：JUC中的Executor框架詳解1
20. 第20天：JUC中的Executor框架詳解2
21. 第21天：java中的CAS，你需要知道的東西
22. 第22天：JUC底層工具類Unsafe，高手必須要了解
23. 第23天：JUC中原子類，一篇就夠了
24. 第24天：ThreadLocal、InheritableThreadLocal（通俗易懂）

java高併發系列連載中，總計估計會有四五十篇文章。

阿里p7一起學併發，公眾號：路人甲java，每天獲取最新文章！

相關推薦