1 ArrayBlockingQueue

ArrayBlockingQueue是一個阻塞隊列，底層使用數(shù)組結(jié)構(gòu)實現(xiàn)，按照先進先出（FIFO）的原則對元素進行排序。

ArrayBlockingQueue是一個線程安全的集合，通過ReentrantLock鎖來實現(xiàn)，在并發(fā)情況下可以保證數(shù)據(jù)的一致性。

此外，ArrayBlockingQueue的容量是有限的，數(shù)組的大小在初始化時就固定了，不會隨著隊列元素的增加而出現(xiàn)擴容的情況，也就是說ArrayBlockingQueue是一個“有界緩存區(qū)”。

在下面圖片中，以數(shù)組形式展示了一個ArrayBlockingQueue：

image

當向隊列插入元素時，首先會插入到數(shù)組的0角標處，再有新元素進來時，依次類推，角標1、角標2、角標3。

整個item[]就是一個隊伍，我們用時間來排序，展示入隊場景。

image

而當有元素出隊時，先移除角標為0的元素，與入隊一樣，依次類推，移除角標1、角標2...上的元素。

這也形成了“先進先出”。

接下來，我們來看看ArrayBlockingQueue的源碼實現(xiàn)！

• 構(gòu)造方法

在多線程中，默認不保證線程公平的訪問隊列。

什么叫做公平訪問隊列？我們都知道，在ArrayBlockingQueue中為了保證數(shù)據(jù)的安全，使用了ReentrantLock鎖。由于鎖的引入，導致了線程之間的競爭。當有一個線程獲取到鎖時，其余線程處于等待狀態(tài)。當鎖被釋放時，所有等待線程為奪鎖而競爭。

而所謂的公平訪問，就是等待的線程在獲取鎖而競爭時，按照等待的先后順序進行獲取操作，先等待的先獲取，后等待的后獲取。

而非公平訪問，就是在獲取時候，無論是先等待還是后等待的線程，均有可能獲取到鎖。

在ArrayBlockingQueue中，由于公平鎖會降低隊列的性能，因而使用非公平鎖(默認)。

是否公平，根據(jù)ReentrantLock對象來實現(xiàn)---ReentrantLock lock = new ReentrantLock(false)，具體看下構(gòu)造便可得知。

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

    //隊列實現(xiàn)：數(shù)組
    final Object[] items;

    //當讀取元素時數(shù)組的下標(下一個被添加元素的索引)
    int takeIndex;

    //添加元素時數(shù)組的下標 （下一個被取出元素的索引）
    int putIndex;

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

    //鎖：
    final ReentrantLock lock;

    //控制take()操作時是否讓線程等待
    private final Condition notEmpty;

    //控制put()操作時是否讓線程等待
    private final Condition notFull;

    //初始化隊列容量構(gòu)造：
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    //帶初始容量大小和公平鎖隊列(公平鎖通過ReentrantLock實現(xiàn))：
    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();
    }
}

• 插入元素

在ArrayBlockingQueue中，提供了兩種不同形式的元素插入--阻塞式和非阻塞式。

對于阻塞式插入來說，當隊列中的元素已滿時，則會將此線程停止，讓其處于等待狀態(tài)，直到隊列中有空余位置產(chǎn)生。

//向隊列尾部添加元素，如果隊列滿了，則線程等待
public void put(E e) throws InterruptedException {
    //不能插入非空元素,會拋出異常
    checkNotNull(e);
    //上鎖，保證數(shù)據(jù)安全
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //隊列中元素 == 數(shù)組長度(隊列滿了),則線程等待
        while (count == items.length)
            notFull.await();
        //添加隊列元素
        insert(e);
    } finally {
        //插入完成，釋放鎖
        lock.unlock();
    }
}

而對于非阻塞式來說，當隊列中的元素已滿時，并不會阻塞此線程的操作，而是讓其返回又或者是拋出異常。

//向隊列尾部添加元素，隊列滿了返回false
public boolean offer(E e) {
    //不能插入非空元素,會拋出異常
    checkNotNull(e);
    //上鎖，保證數(shù)據(jù)安全
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //隊列中元素 == 數(shù)組長度(隊列滿了),則返回false
        if (count == items.length)
            return false;
        else {
            //添加隊列元素
            insert(e);
            return true;
        }
    } finally {
        //插入完成，釋放鎖
        lock.unlock();
    }
}

上面的offer(E e)并不會阻塞線程的執(zhí)行，但是如果想讓阻塞和非阻塞相結(jié)合的話，需要怎么處理?

ArrayBlockingQueue為我們提供了折中的方法--offer(E e, long timeout, TimeUnit unit);

向隊列尾部添加元素，可以設置線程等待時間，如果超過指定時間隊列還是滿的，則返回false；

public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
    //不能插入非空元素,會拋出異常
    checkNotNull(e);
    //轉(zhuǎn)換成超時時間閥值：
    long nanos = unit.toNanos(timeout);
    //上鎖，保證數(shù)據(jù)安全
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        //對隊列是否元素滿了，做判斷。
        while (count == items.length) {
            //如果隊列是滿的，則每次遍歷都去遞減一次nanos的值
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        //添加隊列元素
        insert(e);
        return true;
    } finally {
        //插入完成，釋放鎖
        lock.unlock();
    }
}

以上添加方法，都是通過返回false/true來實現(xiàn)的，而在ArrayBlockingQueue中，還提供了集合最原始的插入方法--add(E e)。

該方法在插入時候，如果隊列中的元素滿了，則會拋出異常。如果插入成功，則返回true。

在add(E e)中，使用父類的add（E e）,實際上其底層也是調(diào)用的offer（E e）方法。

//向隊列尾部添加元素，隊列滿了拋出異常；
public boolean add(E e) {
    return super.add(e);
}

ArrayBlockingQueue中，最底層的插入方法，上面的各種實現(xiàn)，都是基于insert(E x)來實現(xiàn)的。由于insert(E x)是用private來修飾的，所以我們不能直接對其進行調(diào)用。

//插入元素到隊尾，調(diào)整putIndex，喚起等待的獲取線程
private void insert(E x) {
    //向數(shù)組中插入元素
    items[putIndex] = x;
    //設置下一個被取出元素的索引
    putIndex = inc(putIndex);
    //增加隊列元素個數(shù)：
    ++count;
    //喚醒notEmpty上的等待線程
    notEmpty.signal();
}

• 獲取元素

//獲取隊列頭部元素，如果隊列為空，則返回null.不為空。
// 則返回隊列頭部，并從隊列中刪除。
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : extract();
} finally {
lock.unlock();
}
}

    //返回隊列的頭部元素，并從隊列中刪除。如果隊列為空，則等待
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            //如果隊列為空，則進行等待
            while (count == 0)
                notEmpty.await();

            //獲取頭部元素：
            return extract();
        } finally {
            lock.unlock();
        }
    }

    //獲取隊列頭部元素，如果隊列為空，則設置線程等待時間，超過指定時間，還為空，則返回null。
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            return extract();
        } finally {
            lock.unlock();
        }
    }

以上就是關(guān)于ArrayBlockingQueue的全部內(nèi)容！下面，我們繼續(xù)說說LinkedBlockingQueue