多線程（十八、阻塞隊列-ArrayBlockingQueue）

ArrayBlockingQueue簡介

1、隊列的容量一旦在構造時指定，后續不能改變；
2、插入元素時，在隊尾進行；刪除元素時，在隊首進行；
3、隊列滿時，插入元素會阻塞線程；隊列空時，刪除元素也會阻塞線程；
4、支持公平/非公平策略，默認為非公平策略。

ArrayBlockingQueue構造

核心構造方法：

/**
 * 指定隊列初始容量和公平/非公平策略的構造器.
 */
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();  // 當隊列滿時，線程在該隊列等待插入
}

成員變量

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

    /**
     * 內部數組
     */
    final Object[] items;

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

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

    /**
     * 隊列中的元素個數
     */
    int count;

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

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

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

方法

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

插入數據-put

/**
 * 在隊尾插入指定元素，如果隊列已滿，則阻塞線程.
 */
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();
    }
}

入隊方法：enqueue

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

刪除元素-take

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

出隊函數：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;
}

環形結構

1、初始化：

2、插入元素“9”

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

4、插入元素“90”

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

重置代碼看這里：

5、出隊元素“9”

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

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

重置代碼看這里

總結

1、ArrayBlockingQueue利用了ReentrantLock來保證線程的安全性，針對隊列的修改都需要加全局鎖。
2、ArrayBlockingQueue是有界的，且在初始時指定隊列大小。
3、ArrayBlockingQueue的內部數組其實是一種環形結構。