摘要：在隊(duì)尾插入指定元素，如果隊(duì)列已滿，則阻塞線程加鎖隊(duì)列已滿。這里必須用，防止虛假喚醒在隊(duì)列上等待之所以這樣做，是防止線程被意外喚醒，不經(jīng)再次判斷就直接調(diào)用方法。

本文首發(fā)于一世流云專欄：https://segmentfault.com/blog...

ArrayBlockingQueue是在JDK1.5時，隨著J.U.C包引入的一種阻塞隊(duì)列，它實(shí)現(xiàn)了BlockingQueue接口，底層基于數(shù)組實(shí)現(xiàn)：

ArrayBlockingQueue是一種有界阻塞隊(duì)列，在初始構(gòu)造的時候需要指定隊(duì)列的容量。具有如下特點(diǎn)：

隊(duì)列的容量一旦在構(gòu)造時指定，后續(xù)不能改變；

插入元素時，在隊(duì)尾進(jìn)行；刪除元素時，在隊(duì)首進(jìn)行；

隊(duì)列滿時，調(diào)用特定方法插入元素會阻塞線程；隊(duì)列空時，刪除元素也會阻塞線程；

支持公平/非公平策略，默認(rèn)為非公平策略。

這里的公平策略，是指當(dāng)線程從阻塞到喚醒后，以最初請求的順序（FIFO）來添加或刪除元素；非公平策略指線程被喚醒后，誰先搶占到鎖，誰就能往隊(duì)列中添加/刪除順序，是隨機(jī)的。

ArrayBlockingQueue提供了三種構(gòu)造器：

/**
 * 指定隊(duì)列初始容量的構(gòu)造器.
 */
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

/**
 * 指定隊(duì)列初始容量和公平/非公平策略的構(gòu)造器.
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();

    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);     // 利用獨(dú)占鎖的策略
   notEmpty = lock.newCondition();
    notFull = lock.newCondition();
}

/**
 * 根據(jù)已有集合構(gòu)造隊(duì)列
 */
public ArrayBlockingQueue(int capacity, boolean fair, Collection c) {
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock();    // 這里加鎖是用于保證items數(shù)組的可見性
    try {
        int i = 0;
        try {
            for (E e : c) {
                checkNotNull(e);    // 不能有null元素
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;     // 如果隊(duì)列已滿，則重置puIndex索引為0
    } finally {
        lock.unlock();
    }
}

核心就是第二種構(gòu)造器，從構(gòu)造器也可以看出，ArrayBlockingQueue在構(gòu)造時就指定了內(nèi)部數(shù)組的大小，并通過ReentrantLock來保證并發(fā)環(huán)境下的線程安全。

ArrayBlockingQueue的公平/非公平策略其實(shí)就是內(nèi)部ReentrantLock對象的策略，此外構(gòu)造時還創(chuàng)建了兩個Condition對象。在隊(duì)列滿時，插入線程需要在notFull上等待；當(dāng)隊(duì)列空時，刪除線程會在notEmpty上等待：

public class ArrayBlockingQueue extends AbstractQueue
    implements BlockingQueue, java.io.Serializable {

    /**
     * 內(nèi)部數(shù)組
     */
    final Object[] items;

    /**
     * 下一個待刪除位置的索引: take, poll, peek, remove方法使用
     */
    int takeIndex;

    /**
     * 下一個待插入位置的索引: put, offer, add方法使用
     */
    int putIndex;

    /**
     * 隊(duì)列中的元素個數(shù)
     */
    int count;

    /**
     * 全局鎖
     */
    final ReentrantLock lock;

    /**
     * 非空條件隊(duì)列：當(dāng)隊(duì)列空時，線程在該隊(duì)列等待獲取
     */
    private final Condition notEmpty;

    /**
     * 非滿條件隊(duì)列：當(dāng)隊(duì)列滿時，線程在該隊(duì)列等待插入
     */
    private final Condition notFull;

    //...
}

ArrayBlockingQueue會阻塞線程的方法一共4個：put(E e)、offer(e, time, unit)和take()、poll(time, unit)，我們先來看插入元素的方法。

插入元素——put(E e)

插入元素的邏輯很簡單，用ReentrantLock來保證線程安全，當(dāng)隊(duì)列滿時，則調(diào)用線程會在notFull條件隊(duì)列上等待，否則就調(diào)用enqueue方法入隊(duì)。

/**
 * 在隊(duì)尾插入指定元素，如果隊(duì)列已滿，則阻塞線程.
 */
public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();   // 加鎖
    try {
        while (count == items.length)   // 隊(duì)列已滿。這里必須用while，防止虛假喚醒
            notFull.await();            // 在notFull隊(duì)列上等待
        enqueue(e);                     // 隊(duì)列未滿, 直接入隊(duì)
    } finally {
        lock.unlock();
    }
}

這里需要注意一點(diǎn)，隊(duì)列已滿的時候，是通過while循環(huán)判斷的,這其實(shí)是多線程設(shè)計模式中的Guarded Suspension模式：

while (count == items.length)   // 隊(duì)列已滿。這里必須用while，防止虛假喚醒
    notFull.await();            // 在notFull隊(duì)列上等待

之所以這樣做，是防止線程被意外喚醒，不經(jīng)再次判斷就直接調(diào)用enqueue方法。

enqueue方法：

private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)     // 隊(duì)列已滿,則重置索引為0
        putIndex = 0;
    count++;                            // 元素個數(shù)+1
    notEmpty.signal();                  // 喚醒一個notEmpty上的等待線程(可以來隊(duì)列取元素了)
}

刪除元素——take()

刪除元素的邏輯和插入元素類似，區(qū)別就是：刪除元素時，如果隊(duì)列空了，則線程需要在notEmpty條件隊(duì)列上等待。

/**
 * 從隊(duì)首刪除一個元素, 如果隊(duì)列為空, 則阻塞線程
 */
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)      // 隊(duì)列為空, 則線程在notEmpty條件隊(duì)列等待
            notEmpty.await();
        return dequeue();       // 隊(duì)列非空，則出隊(duì)一個元素
    } finally {
        lock.unlock();
    }
}

隊(duì)列非空時，調(diào)用dequeue方法出隊(duì)一個元素：

private E dequeue() {
    final Object[] items = this.items;
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)    // 如果隊(duì)列已空
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();                   // 喚醒一個notFull上的等待線程(可以插入元素到隊(duì)列了)
    return x;
}

從上面的入隊(duì)/出隊(duì)操作，可以看出，ArrayBlockingQueue的內(nèi)部數(shù)組其實(shí)是一種環(huán)形結(jié)構(gòu)。

假設(shè)ArrayBlockingQueue的容量大小為6，我們來看下整個入隊(duì)過程：

①初始時

②插入元素“9”

③插入元素“2”、“10”、“25”、“93”

④插入元素“90”

注意，此時再插入一個元素“90”，則putIndex變成6，等于隊(duì)列容量6，由于是循環(huán)隊(duì)列，所以會將tableIndex重置為0：

這是隊(duì)列已經(jīng)滿了（count==6），如果再有線程嘗試插入元素，并不會覆蓋原有值，而是被阻塞。

我們再來看下出隊(duì)過程：

①出隊(duì)元素“9”

②出隊(duì)元素“2”、“10”、“25”、“93”

③出隊(duì)元素“90”

注意，此時再出隊(duì)一個元素“90”，則tabeIndex變成6，等于隊(duì)列容量6，由于是循環(huán)隊(duì)列，所以會將tableIndex重置為0：

這是隊(duì)列已經(jīng)空了（count==0），如果再有線程嘗試出隊(duì)元素，則會被阻塞。

ArrayBlockingQueue利用了ReentrantLock來保證線程的安全性，針對隊(duì)列的修改都需要加全局鎖。在一般的應(yīng)用場景下已經(jīng)足夠。對于超高并發(fā)的環(huán)境，由于生產(chǎn)者-消息者共用一把鎖，可能出現(xiàn)性能瓶頸。

另外，由于ArrayBlockingQueue是有界的，且在初始時指定隊(duì)列大小，所以如果初始時需要限定消息隊(duì)列的大小，則ArrayBlockingQueue 比較合適。后續(xù)，我們會介紹另一種基于單鏈表實(shí)現(xiàn)的阻塞隊(duì)列——LinkedBlockingQueue，該隊(duì)列的最大特點(diǎn)是使用了“兩把鎖”，以提升吞吐量。