3.12 ArrayBlockingQueue詳解

3.12 ArrayBlockingQueue

ArrayBlockingQueue是阻塞佇列BolockingQueue介面基于陣列的實作，該特點由名字也可看出，

3.12.1 ArrayBlockingQueue繼承關系

ArrayBlockingQueue繼承了抽象類AbstractQueue并且實作了BlockingQueue介面，

3.12.2 ArrayBlockingQueue用法

創建一個ArrayBlockingQueue，必須要指定該佇列長度，注意該佇列長度是固定的，所以分配適當的長度非常重要，視具體的需求而定，

BlockingQueue<Integer> queue = new ArrayBlockingQueue(1024);

入隊方法offer，還可以使用入隊的其他方法

queue.offer(1);
queue.offer(3);

出隊poll，還有其他方法入take

Integer poll = queue.poll();
try {
    Integer take = queue.take();
} catch (InterruptedException e) {
    e.printStackTrace();
}

3.12.3 ArrayBlockingQueue內部組成（原始碼決議）

/** The queued items 佇列存放元素的陣列 */
final Object[] items;
/** items index for next take, poll, peek or remove 出隊索引位置*/
int takeIndex;
/** items index for next put, offer, or add 入隊索引位置*/
int putIndex;
/** Number of elements in the queue 佇列中元素個數 */
int count;
/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook. 下邊的用于并發控制
 */
/** Main lock guarding all access 可重入鎖*/
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;

ArrayBlockingQueue中存放元素的陣列為一個Object[] items的陣列，另外有一個ReentrantLock lock用于并發控制，Codition notEmpty是一個ConiditonObject類物件，內部維護了一個CLH鏈表，該聊表維護的是一個等待執行緒之間的關系，其原始碼可在AbstractQueuedSynchronizer抽象類中找到，

3.12.4 ArrayBlockingQueue構造方法

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
//初始化容器陣列和鎖和CLH佇列
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 ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        int i = 0;
        try {
            for (E e : c) {
                checkNotNull(e);
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

3.12.5 ArrayBlockingQueue入隊和出隊

offer方法，和一般佇列并沒有區別，只是加了鎖，鎖的釋放一定在finally代碼塊中，保證鎖的釋放，

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
            return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

出隊方法offer和take，注意二者的區別，take方法當佇列為空的時候執行緒會等待，而poll方法在佇列為空的時候直接回傳null，對于阻塞佇列如該佇列長時間為空的情況下使用take方法是很浪費資源的，

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
        	//注意這一行代碼
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

總結ArrayBlockingQueue是阻塞佇列基于陣列的實作，需要注意的是內部使用的鎖是ReentrantLock，并且了解不同出隊方法的區別，