多線程（十八、阻塞隊(duì)列-ArrayBlockingQueue）

ArrayBlockingQueue簡介

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

ArrayBlockingQueue構(gòu)造

核心構(gòu)造方法：

/**
 * 指定隊(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();  //當(dāng)隊(duì)列空時，線程在該隊(duì)列等待獲取
    notFull = lock.newCondition();  // 當(dāng)隊(duì)列滿時，線程在該隊(duì)列等待插入
}

成員變量

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, 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)，我們先來看插入元素的方法。

插入數(shù)據(jù)-put

/**
 * 在隊(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();
    }
}

入隊(duì)方法：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

/**
 * 從隊(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ì)函數(shù)：dequeue

/**
     *刪除一個元素，則喚醒一個等待插入的線程
     */
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;
}

環(huán)形結(jié)構(gòu)

1、初始化：

2、插入元素“9”

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

4、插入元素“90”

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

重置代碼看這里：

5、出隊(duì)元素“9”

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

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

重置代碼看這里

總結(jié)

1、ArrayBlockingQueue利用了ReentrantLock來保證線程的安全性，針對隊(duì)列的修改都需要加全局鎖。
2、ArrayBlockingQueue是有界的，且在初始時指定隊(duì)列大小。
3、ArrayBlockingQueue的內(nèi)部數(shù)組其實(shí)是一種環(huán)形結(jié)構(gòu)。