ArrayBlockingQueue底層是用數(shù)組實現(xiàn)的有界(即大小固定)的FIFO隊列,其中利用兩個字段,將該數(shù)組構(gòu)造成了環(huán)形數(shù)組。ArrayBlockingQueue體現(xiàn)了生產(chǎn)者-消費者模型,通過ReentrantLock和Condition實現(xiàn)了資源的互斥訪問和線程間的通信,保證了線程安全。
1. ArrayBlockingQueue繼承關(guān)系圖

2. ArrayBlockingQueue源碼分析
2.1 字段
// 這4個字段無需被volatile修飾,因為鎖可以保證它們的可見性
// 底層數(shù)組
final Object[] items;
// 利用takeIndex和putIndex將數(shù)組構(gòu)造成了環(huán)形數(shù)組
// takeIndex表示下次take、poll、peek、remove時元素的索引
// (即環(huán)形數(shù)組中第一個元素的索引)
int takeIndex;
// putIndex表示下次put、offer、add時元素的索引
// (即環(huán)形數(shù)組中最后一個元素下一個位置的索引)
int putIndex;
// 隊列中元素的個數(shù)
int count;
// 鎖
final ReentrantLock lock;
// 出隊、入隊時要用的兩個Condition(經(jīng)典雙條件算法)
private final Condition notEmpty;
private final Condition notFull;
2.2 三個構(gòu)造方法
(1)ArrayBlockingQueue(int,boolean)
public ArrayBlockingQueue(int capacity, boolean fair) {
// 檢查容量
if (capacity <= 0)
throw new IllegalArgumentException();
// 創(chuàng)建指定大小的數(shù)組
this.items = new Object[capacity];
// fair為true,表示創(chuàng)建公平鎖,為false,表示創(chuàng)建非公平鎖
// 公平性通常會降低吞吐量,但可以避免饑餓
lock = new ReentrantLock(fair);
// 創(chuàng)建基于ReentrantLock的兩個Condition
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
(2)ArrayBlockingQueue(int)
public ArrayBlockingQueue(int capacity) {
// 調(diào)用ArrayBlockingQueue(int,boolean)
// 第二個參數(shù)為false,表示使用非公平鎖
this(capacity, false);
}
(3)ArrayBlockingQueue(int,boolean,Collection)
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
// 調(diào)用ArrayBlockingQueue(int,boolean)初始化各個字段
this(capacity, fair);
// 不會出現(xiàn)多個線程同時構(gòu)造一個ArrayBlockingQueue
// 對象,所以這里加鎖是為了保證可見性,而不是互斥性
final ReentrantLock lock = this.lock;
lock.lock();
try {
int i = 0;
try {
// 將c中元素添加到items中
for (E e : c) {
checkNotNull(e); // e不能為null
items[i++] = e;
}
// 若c中元素個數(shù)超過items的大小,拋出越界異常
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i; // 初始化count
// 更新putIndex,takeIndex默認(rèn)為0
// 注意putIndex是環(huán)形數(shù)組中最后一個元素下一個位置
// 的索引,takeIndex是環(huán)形數(shù)組中第一個元素的索引
putIndex = (i == capacity) ? 0 : i;
} finally {
// 解鎖
lock.unlock();
}
}
2.3 輔助方法
(1)enqueue
// 入隊
private void enqueue(E x) {
final Object[] items = this.items;
// 將x存至putIndex處
items[putIndex] = x;
// 檢查、維護(hù)環(huán)形數(shù)組
if (++putIndex == items.length)
putIndex = 0;
// 更新count
count++;
// 喚醒一個通過notEmpty.await或notEmpty.awaitNanos阻塞著的消費者線程
// 其實調(diào)用notEmpty.signal后只是將該消費者線程對應(yīng)的節(jié)點從notEmpty
// 中維護(hù)的條件隊列轉(zhuǎn)移到了ReentrantLock維護(hù)的同步隊列中,當(dāng)某個線
// 程釋放鎖且喚醒的是該消費者線程時,該消費者線程才算被真正喚醒
notEmpty.signal();
}
(2)dequeue
// 出隊
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
// 從takeIndex處取出元素
E x = (E) items[takeIndex];
items[takeIndex] = null;
// 檢查、維護(hù)環(huán)形數(shù)組
if (++takeIndex == items.length)
takeIndex = 0;
count--; // 更新count
// itrs相關(guān)(略)
// 喚醒一個通過notFull.await或notFull.awaitNanos阻塞著的生產(chǎn)者線程
notFull.signal();
return x;
}
2.4 添加元素
(1)put-無限期阻塞版
public void put(E e) throws InterruptedException {
checkNotNull(e); // e不能為null
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 會處理中斷的鎖
try {
//ArrayBlockingQueue的put、take等方法中用while的原因(通過兩個例子進(jìn)行說明):
// 例1:非公平鎖、隊列容量為2時的情形:
// 假設(shè)有三個線程thread0、thread1、thread2,初始時,thread0調(diào)用了兩次put添加了
// 兩個元素,之后thread1獲取到鎖調(diào)用put后會阻塞(因為隊列已滿),之后thread2獲取
// 到鎖,調(diào)用了take取出一個元素并調(diào)用了notFull.signal后釋放鎖,之后thread0先獲
// 取到鎖調(diào)用了一次put后釋放鎖,之后thread1才獲取到鎖,此時隊列仍是滿的(while
// 中的條件再次成立),因此thread1會繼續(xù)阻塞,所以必須得用while
// 例2:公平鎖、隊列容量為2時的情形:
// 將上面用非公平鎖的例子改為:thread0第三次put時,thread2還未釋放鎖且還未調(diào)用
// notFull.signal,之后thread0對應(yīng)的Node先加入同步隊列,之后thread2調(diào)用notFull.signal
// 將thread1對應(yīng)的Node加入到同步隊列,之后thread2釋放鎖,先喚醒thread0,被喚醒
// 的thread0添加完元素后釋放鎖,喚醒thread1,此時隊列仍然是滿的,while中的條件
// 再次成立,因此thread1會繼續(xù)阻塞,所以必須得用while
// 隊列已滿,無法放入新元素
while (count == items.length)
// 阻塞等待被其他消費者線程喚醒
notFull.await();
// 入隊
enqueue(e);
} finally {
// 解鎖
lock.unlock();
}
}
(2)add-拋異常版
public boolean add(E e) {
// 調(diào)用AbstractQueue.add
return super.add(e);
}
// AbstractQueue.add
public boolean add(E e) {
// offer返回true,表示(隊列未滿)添加成功,
// 返回false,表示(隊列已滿)添加失敗
if (offer(e))
return true;
else // 添加失敗,拋出異常
throw new IllegalStateException("Queue full");
}
(3)offer(E)-非阻塞版
public boolean offer(E e) {
checkNotNull(e); // e不能為null
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 隊列已滿,直接返回false
if (count == items.length)
return false;
else { // 隊列未滿
// 入隊
enqueue(e);
// 入隊成功,返回true
return true;
}
} finally {
// 解鎖
lock.unlock();
}
}
(4)offer(E,long,TimeUnit)-限期阻塞版
// timeout是時長,TimeUnit是時間單位
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
checkNotNull(e); // e不能為null
// 轉(zhuǎn)化為納秒
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 隊列已滿
while (count == items.length) {
// 第二次到這里說明while中的條件仍不成立,隊列
// 仍然是滿的,已經(jīng)達(dá)到阻塞時長,直接返回false
if (nanos <= 0)
return false;
// 阻塞指定時長這種情況下if中的條件會不成立
// (也可能被notFull.signal提前喚醒,這種情況下while中的條件會不成立)
// awaitNanos返回的是(粗略計算):
// (阻塞前的系統(tǒng)時間+nanos)-阻塞結(jié)束后的系統(tǒng)時間
nanos = notFull.awaitNanos(nanos);
}
// 入隊
enqueue(e);
// 入隊成功,返回true
return true;
} finally {
// 解鎖
lock.unlock();
}
}
2.5 獲取(刪除)元素
(1)take-無限期阻塞版
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 隊列為空
while (count == 0)
// 阻塞等待被其他消生產(chǎn)者線程喚醒
notEmpty.await();
// 出隊
return dequeue();
} finally {
// 解鎖
lock.unlock();
}
}
(2)poll()-非阻塞版
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 隊列為空直接返回null
return (count == 0) ? null : dequeue();
} finally {
// 解鎖
lock.unlock();
}
}
(3)poll(long,TimeUnit)-限期阻塞版
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
// 轉(zhuǎn)化為納秒
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
// 與offer(E,long,TimeUnit)是相對的
// 隊列為空
while (count == 0) {
if (nanos <= 0)
return null;
// 阻塞指定時長或被notEmpty.signal提前喚醒
nanos = notEmpty.awaitNanos(nanos);
}
// 出隊
return dequeue();
} finally {
// 解鎖
lock.unlock();
}
}
(4)remove-刪除指定元素
// 注意take、poll()、poll(long,TimeUnit)的返回值類型都是E,外部可接收從隊列中
// 移除的元素,而remove則是直接刪除隊列中第一個滿足o.equals(items[i])的元素
public boolean remove(Object o) {
// o為null直接返回false,因為隊列中允許存在為null的元素
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 因為用的do-while,所以這里必須先判斷count是否大于0
if (count > 0) {
final int putIndex = this.putIndex;
int i = takeIndex;
// 強調(diào):putIndex是環(huán)形數(shù)組最后一個元素后一個位置的索引
// 刪除takeIndex到putIndex-1之間第一個滿足o.equals(items[i])的元素
do {
if (o.equals(items[i])) {
// 刪除i處的元素,并將i+1到putIndex-1之間的所有元素向前移動
removeAt(i);
// 已經(jīng)將一個滿足的移除,返回true
return true;
}
// 檢查、維護(hù)環(huán)形數(shù)組
if (++i == items.length)
i = 0;
} while (i != putIndex);
}
// 不存在滿足o.equals(items[i])的元素,返回false
return false;
} finally {
// 解鎖
lock.unlock();
}
}
void removeAt(final int removeIndex) {
final Object[] items = this.items;
// 這里算是一個小優(yōu)化:
// 當(dāng)removeIndex是takeIndex時,僅將takeIndex加1,否則在else
// 中移動putIndex,將removeIndex ~ putIndex的所有元素向前移動
if (removeIndex == takeIndex) {
items[takeIndex] = null;
// 檢查、維護(hù)環(huán)形數(shù)組
if (++takeIndex == items.length)
takeIndex = 0;
count--;
// itrs相關(guān)(略)
} else {
final int putIndex = this.putIndex;
// i初始值為removeIndex
for (int i = removeIndex;;) {
int next = i + 1;
// 檢查、維護(hù)環(huán)形數(shù)組
if (next == items.length)
next = 0;
// next不為putIndex
if (next != putIndex) {
items[i] = items[next];
// 用next更新i
// 因為是環(huán)形數(shù)組,所有不能直接i++
i = next;
} else {
items[i] = null;
// 因為是環(huán)形數(shù)組,所有不能直接this.putIndex++
this.putIndex = i;
break; // 移動完畢,跳出循環(huán)
}
}
count--;
// itrs相關(guān)(略)
}
// 喚醒一個通過notFull.await或notFull.awaitNanos阻塞著的生產(chǎn)者線程
notFull.signal();
}
2.6 查看元素
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 返回item[takeIndex]
// 當(dāng)隊列為空時,item[takeIndex]為null
return itemAt(takeIndex); // null when queue is empty
} finally {
// 解鎖
lock.unlock();
}
}
final E itemAt(int i) {
return (E) items[i];
}