Java多線程進(jìn)階（三二）—— J.U.C之collections框架：ArrayBlockingQ

entner 發(fā)布于2019-08-19 11:07 / 720人閱讀

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

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

一、ArrayBlockingQueue簡介

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

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

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

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

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

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

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

二、ArrayBlockingQueue原理 構(gòu)造

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

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

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

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

/**
 * 根據(jù)已有集合構(gòu)造隊列
 */
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;     // 如果隊列已滿，則重置puIndex索引為0
    } finally {
        lock.unlock();
    }
}

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

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

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

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

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

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

    //...
}

核心方法

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

插入元素——put(E e)

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

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

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

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

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

enqueue方法：

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

刪除元素——take()

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

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

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

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

環(huán)形隊列

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

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

①初始時

②插入元素“9”

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

④插入元素“90”

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

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

我們再來看下出隊過程：

①出隊元素“9”

②出隊元素“2”、“10”、“25”、“93”

③出隊元素“90”

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

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

三、總結(jié)

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

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