Java併發（10）- 簡單聊聊JDK中的七大阻塞佇列
引言
JDK中除了上文提到的各種併發容器，還提供了豐富的阻塞佇列。阻塞佇列統一實現了BlockingQueue
//插入元素，佇列滿後會丟擲異常 boolean add(E e); //移除元素，佇列為空時會丟擲異常 E remove(); //插入元素，成功反會true boolean offer(E e); //移除元素 E poll(); //插入元素，佇列滿後會阻塞 void put(E e) throws InterruptedException; //移除元素，佇列空後會阻塞 E take() throws InterruptedException; //限時插入 boolean offer(E e, long timeout, TimeUnit unit) //限時移除 E poll(long timeout, TimeUnit unit); //獲取所有元素到Collection中 int drainTo(Collection<? super E> c);
JDK1.8中的阻塞佇列實現共有7個，分別是ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue、SynchronousQueue、LinkedTransferQueue以及LinkedBlockingDeque，下面就來一一對他們進行一個簡單的分析。
ArrayBlockingQueue
ArrayBlockingQueue是一個底層用陣列實現的有界阻塞佇列，有界是指他的容量大小是固定的，不能擴充容量，在初始化時就必須確定佇列大小。它通過可重入的獨佔鎖ReentrantLock來控制併發，Condition來實現阻塞。
//通過陣列來儲存佇列中的元素 final Object[] items; //初始化一個固定的陣列大小，預設使用非公平鎖來控制併發 public ArrayBlockingQueue(int capacity) { this(capacity, false); } //初始化固定的items陣列大小，初始化notEmpty以及notFull兩個Condition來控制生產消費 public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair);//通過ReentrantLock來控制併發 notEmpty = lock.newCondition(); notFull = lock.newCondition(); }
可以看到ArrayBlockingQueue初始化了一個ReentrantLock以及兩個Condition，用來控制併發下佇列的生產消費。這裡重點看下阻塞的put以及take方法：
//插入元素到佇列中
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); //獲取獨佔鎖
try {
while (count == items.length) //如果佇列已滿則通過await阻塞put方法
notFull.await();
enqueue(e); //插入元素
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length) //插入元素後將putIndex+1，當佇列使用完後重置為0
putIndex = 0;
count++;
notEmpty.signal(); //佇列新增元素後喚醒因notEmpty等待的消費執行緒
}
//移除佇列中的元素
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); //獲取獨佔鎖
try {
while (count == 0) //如果佇列已空則通過await阻塞take方法
notEmpty.await();
return dequeue(); //移除元素
} finally {
lock.unlock();
}
}
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length) //移除元素後將takeIndex+1，當佇列使用完後重置為0
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal(); //佇列消費元素後喚醒因notFull等待的消費執行緒
return x;
}在佇列新增和移除元素的過程中使用putIndex、takeIndex以及count三個變數來控制生產消費元素的過程，putIndex負責記錄下一個可新增元素的下標，takeIndex負責記錄下一個可移除元素的下標，count記錄了佇列中的元素總量。佇列滿後通過notFull.await()來阻塞生產者執行緒，消費元素後通過notFull.signal()來喚醒阻塞的生產者執行緒。佇列為空後通過notEmpty.await()來阻塞消費者執行緒，生產元素後通過notEmpty.signal()喚醒阻塞的消費者執行緒。
限時插入以及移除方法在ArrayBlockingQueue中通過awaitNanos來實現，在給定的時間過後如果執行緒未被喚醒則直接返回。
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
checkNotNull(e);
long nanos = unit.toNanos(timeout); //獲取定時時長
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
if (nanos <= 0) //指定時長過後，執行緒仍然未被喚醒則返回false
return false;
nanos = notFull.awaitNanos(nanos); //指定時長內阻塞執行緒
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}還有一個比較重要的方法：drainTo，drainTo方法可以一次性獲取佇列中所有的元素，它減少了鎖定佇列的次數，使用得當在某些場景下對效能有不錯的提升。
public int drainTo(Collection<? super E> c, int maxElements) {
checkNotNull(c);
if (c == this)
throw new IllegalArgumentException();
if (maxElements <= 0)
return 0;
final Object[] items = this.items;
final ReentrantLock lock = this.lock; //僅獲取一次鎖
lock.lock();
try {
int n = Math.min(maxElements, count); //獲取佇列中所有元素
int take = takeIndex;
int i = 0;
try {
while (i < n) {
@SuppressWarnings("unchecked")
E x = (E) items[take];
c.add(x); //迴圈插入元素
items[take] = null;
if (++take == items.length)
take = 0;
i++;
}
return n;
} finally {
// Restore invariants even if c.add() threw
if (i > 0) {
count -= i;
takeIndex = take;
if (itrs != null) {
if (count == 0)
itrs.queueIsEmpty();
else if (i > take)
itrs.takeIndexWrapped();
}
for (; i > 0 && lock.hasWaiters(notFull); i--)
notFull.signal(); //喚醒等待的生產者執行緒
}
}
} finally {
lock.unlock();
}
}LinkedBlockingQueue
LinkedBlockingQueue是一個底層用單向連結串列實現的有界阻塞佇列，和ArrayBlockingQueue一樣，採用ReentrantLock來控制併發，不同的是它使用了兩個獨佔鎖來控制消費和生產。put以及take方法原始碼如下：
public void put(E e) throws InterruptedException {
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
//因為使用了雙鎖，需要使用AtomicInteger計算元素總量，避免併發計算不準確
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
notFull.await(); //佇列已滿，阻塞生產執行緒
}
enqueue(node); //插入元素到佇列尾部
c = count.getAndIncrement(); //count + 1
if (c + 1 < capacity) //如果+1後佇列還未滿，通過其他生產執行緒繼續生產
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0) //只有當之前是空時，消費佇列才會阻塞，否則是不需要通知的
signalNotEmpty();
}
private void enqueue(Node<E> node) {
//將新元素新增到連結串列末尾，然後將last指向尾部元素
last = last.next = node;
}
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await(); //佇列為空，阻塞消費執行緒
}
x = dequeue(); //消費一個元素
c = count.getAndDecrement(); //count - 1
if (c > 1) // 通知其他等待的消費執行緒繼續消費
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity) //只有當之前是滿的，生產佇列才會阻塞，否則是不需要通知的
signalNotFull();
return x;
}
//消費佇列頭部的下一個元素，同時將新頭部置空
private E dequeue() {
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;
}可以看到LinkedBlockingQueue通過takeLock和putLock兩個鎖來控制生產和消費，互不干擾，只要佇列未滿，生產執行緒可以一直生產，只要佇列不為空，消費執行緒可以一直消費，不會相互因為獨佔鎖而阻塞。
看過了LinkedBlockingQueue以及ArrayBlockingQueue的底層實現，會發現一個問題，正常來說消費者和生產者可以併發執行對佇列的吞吐量會有比較大的提升，那麼為什麼ArrayBlockingQueue中不使用雙鎖來實現佇列的生產和消費呢？我的理解是ArrayBlockingQueue也能使用雙鎖來實現功能，但由於它底層使用了陣列這種簡單結構，相當於一個共享變數，如果通過兩個鎖，需要更加精確的鎖控制，這也是為什麼JDK1.7中的ConcurrentHashMap使用了分段鎖來實現，將一個數組分為多個數組來提高併發量。LinkedBlockingQueue不存在這個問題，連結串列這種資料結構頭尾節點都相對獨立，儲存上也不連續，雙鎖控制不存在複雜性。這是我的理解，如果你有更好的結論，請留言探討。
PriorityBlockingQueue
PriorityBlockingQueue是一個底層由陣列實現的無界佇列，並帶有排序功能，同樣採用ReentrantLock來控制併發。由於是無界的，所以插入元素時不會阻塞，沒有佇列滿的狀態，只有佇列為空的狀態。通過這兩點特徵其實可以猜測它應該是有一個獨佔鎖（底層陣列）和一個Condition（只通知消費）來實現的。put以及take方法原始碼分析如下：
public void put(E e) {
offer(e);
}
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
//無界佇列，佇列長度不夠時會擴容
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
//通過comparator來實現優先順序排序
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal(); //和ArrayBlockingQueue一樣，每次新增元素後通知消費執行緒
} finally {
lock.unlock();
}
return true;
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ( (result = dequeue()) == null)
notEmpty.await(); //佇列為空，阻塞消費執行緒
} finally {
lock.unlock();
}
return result;
}DelayQueue
DelayQueue也是一個無界佇列，它是在PriorityQueue基礎上實現的，先按延遲優先順序排序，延遲時間短的排在前面。和PriorityBlockingQueue相似，底層也是陣列，採用一個ReentrantLock來控制併發。由於是無界的，所以插入元素時不會阻塞，沒有佇列滿的狀態。能想到的最簡單的使用場景一般有兩個：一個是快取過期，一個是定時執行的任務。但由於是無界的，快取過期上一般使用的並不多。簡單來看下put以及take方法：
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();//優先順序佇列
public void put(E e) {
offer(e);
}
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e); //插入元素到優先順序佇列
if (q.peek() == e) { //如果插入的元素在佇列頭部
leader = null;
available.signal(); //通知消費執行緒
}
return true;
} finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek(); //獲取頭部元素
if (first == null)
available.await(); //空佇列阻塞
else {
long delay = first.getDelay(NANOSECONDS); //檢查元素是否延遲到期
if (delay <= 0)
return q.poll(); //到期則彈出元素
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay); //阻塞未到期的時間
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}SynchronousQueue
SynchronousQueue相比較之前的4個佇列就比較特殊了，它是一個沒有容量的佇列，也就是說它內部時不會對資料進行儲存，每進行一次put之後必須要進行一次take，否則相同執行緒繼續put會阻塞。這種特性很適合做一些傳遞性的工作，一個執行緒生產，一個執行緒消費。內部分為公平和非公平訪問兩種模式，預設使用非公平，未使用鎖，全部通過CAS操作來實現併發，吞吐量非常高。這裡只對它的非公平實現下的take和put方法做下簡單分析：
//非公平情況下呼叫內部類TransferStack的transfer方法put
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
if (transferer.transfer(e, false, 0) == null) {
Thread.interrupted();
throw new InterruptedException();
}
}
//非公平情況下呼叫內部類TransferStack的transfer方法take
public E take() throws InterruptedException {
E e = transferer.transfer(null, false, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
//具體的put以及take方法，只有E的區別，通過E來區別REQUEST還是DATA模式
E transfer(E e, boolean timed, long nanos) {
SNode s = null; // constructed/reused as needed
int mode = (e == null) ? REQUEST : DATA;
for (;;) {
SNode h = head;
//棧無元素或者元素和插入的元素模式相匹配，也就是說都是插入元素
if (h == null || h.mode == mode) {
//有時間限制並且超時
if (timed && nanos <= 0) {
if (h != null && h.isCancelled())
casHead(h, h.next); // 重新設定頭節點
else
return null;
}
//未超時cas操作嘗試設定頭節點
else if (casHead(h, s = snode(s, e, h, mode))) {
//自旋一段時間後未消費元素則掛起put執行緒
SNode m = awaitFulfill(s, timed, nanos);
if (m == s) { // wait was cancelled
clean(s);
return null;
}
if ((h = head) != null && h.next == s)
casHead(h, s.next); // help s's fulfiller
return (E) ((mode == REQUEST) ? m.item : s.item);
}
}
//棧不為空並且和頭節點模式不匹配，存在元素則消費元素並重新設定head節點
else if (!isFulfilling(h.mode)) { // try to fulfill
if (h.isCancelled()) // already cancelled
casHead(h, h.next); // pop and retry
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
for (;;) { // loop until matched or waiters disappear
SNode m = s.next; // m is s's match
if (m == null) { // all waiters are gone
casHead(s, null); // pop fulfill node
s = null; // use new node next time
break; // restart main loop
}
SNode mn = m.next;
if (m.tryMatch(s)) {
casHead(s, mn); // pop both s and m
return (E) ((mode == REQUEST) ? m.item : s.item);
} else // lost match
s.casNext(m, mn); // help unlink
}
}
}
//節點正在匹配階段
else { // help a fulfiller
SNode m = h.next; // m is h's match
if (m == null) // waiter is gone
casHead(h, null); // pop fulfilling node
else {
SNode mn = m.next;
if (m.tryMatch(h)) // help match
casHead(h, mn); // pop both h and m
else // lost match
h.casNext(m, mn); // help unlink
}
}
}
}
//先自旋後掛起的核心方法
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
//計算自旋的次數
int spins = (shouldSpin(s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel();
SNode m = s.match;
//匹配成功過返回節點
if (m != null)
return m;
//超時控制
if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
s.tryCancel();
continue;
}
}
//自旋檢查，是否進行下一次自旋
if (spins > 0)
spins = shouldSpin(s) ? (spins-1) : 0;
else if (s.waiter == null)
s.waiter = w; // establish waiter so can park next iter
else if (!timed)
LockSupport.park(this); //在這裡掛起執行緒
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}程式碼非常複雜，這裡說下我所理解的核心邏輯。程式碼中可以看到put以及take方法都是通過呼叫transfer方法來實現的，然後通過引數mode來區別，在生產元素時如果是同一個執行緒多次put則會採取自旋的方式多次嘗試put元素，可能自旋過程中元素會被消費，這樣可以及時put，降低執行緒掛起的效能損耗，高吞吐量的核心也在這裡，消費執行緒一樣，空棧時也會先自旋，自旋失敗然後通過執行緒的LockSupport.park方法掛起。
LinkedTransferQueue
LinkedTransferQueue是一個無界的阻塞佇列，底層由連結串列實現。雖然和LinkedBlockingQueue一樣也是連結串列實現的，但併發控制的實現上卻很不一樣，和SynchronousQueue類似，採用了大量的CAS操作，沒有使用鎖，由於是無界的，所以不會put生產執行緒不會阻塞，只會在take時阻塞消費執行緒，消費執行緒掛起時同樣使用LockSupport.park方法。
LinkedTransferQueue相比於以上的佇列還提供了一些額外的功能，它實現了TransferQueue介面，有兩個關鍵方法transfer(E e)和tryTransfer(E e)方法，transfer在沒有消費時會阻塞，tryTransfer在沒有消費時不會插入到佇列中，也不會等待，直接返回false。
private static final int NOW = 0; // for untimed poll, tryTransfer
private static final int ASYNC = 1; // for offer, put, add
private static final int SYNC = 2; // for transfer, take
private static final int TIMED = 3; // for timed poll, tryTransfer
//通過SYNC狀態來實現生產阻塞
public void transfer(E e) throws InterruptedException {
if (xfer(e, true, SYNC, 0) != null) {
Thread.interrupted(); // failure possible only due to interrupt
throw new InterruptedException();
}
}
//通過NOW狀態跳過新增元素以及阻塞
public boolean tryTransfer(E e) {
return xfer(e, true, NOW, 0) == null;
}
//通過ASYNC狀態跳過阻塞
public void put(E e) {
xfer(e, true, ASYNC, 0);
}
//通過SYNC狀態來實現消費阻塞
public E take() throws InterruptedException {
E e = xfer(null, false, SYNC, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
//生產消費呼叫同一個方法，通過e是否為空，haveData，how等引數來區分具體邏輯
private E xfer(E e, boolean haveData, int how, long nanos) {
if (haveData && (e == null))
throw new NullPointerException();
Node s = null; // the node to append, if needed
retry:
for (;;) { // restart on append race
//找出第一個可用節點
for (Node h = head, p = h; p != null;) { // find & match first node
boolean isData = p.isData;
Object item = p.item;
//佇列為空時直接跳過
if (item != p && (item != null) == isData) { // unmatched
//節點型別相同，跳過
if (isData == haveData) // can't match
break;
if (p.casItem(item, e)) { // match
for (Node q = p; q != h;) {
Node n = q.next; // update by 2 unless singleton
if (head == h && casHead(h, n == null ? q : n)) {
h.forgetNext();
break;
} // advance and retry
if ((h = head) == null ||
(q = h.next) == null || !q.isMatched())
break; // unless slack < 2
}
LockSupport.unpark(p.waiter);
return LinkedTransferQueue.<E>cast(item);
}
}
Node n = p.next;
p = (p != n) ? n : (h = head); // Use head if p offlist
}
//插入節點或移除節點具體邏輯
//tryTransfer方法會直接跳過並返回結果
if (how != NOW) { // No matches available
if (s == null)
s = new Node(e, haveData);
Node pred = tryAppend(s, haveData); //加入節點
if (pred == null)
continue retry; // lost race vs opposite mode
if (how != ASYNC)
//自旋以及阻塞消費執行緒邏輯，和SynchronousQueue類似，先嚐試自選，失敗後掛起執行緒
//transfer方法在沒有消費執行緒時也會阻塞在這裡
return awaitMatch(s, pred, e, (how == TIMED), nanos);
}
return e; // not waiting
}
}LinkedBlockingDeque
LinkedBlockingDeque是一個有界的雙端佇列，底層採用一個雙向的連結串列來實現，在LinkedBlockingQeque的Node
總結
文章對JDK1.8中的7種阻塞佇列都做了簡單分析，幫助大家大致梳理的這7個佇列的基本原理。總的來說每種阻塞佇列都有它自己的應用場景，使用時可以先根據有界還是無界，然後在根據各自的特性來進行選擇。
有界阻塞佇列包括：ArrayBlockingQueue、LinkedBlockingQueue以及LinkedBlockingDeque三種，LinkedBlockingDeque應用場景很少，一般用在“工作竊取”模式下。ArrayBlockingQueue和LinkedBlockingQueue基本就是陣列和連結串列的區別。無界佇列包括PriorityBlockingQueue、DelayQueue和LinkedTransferQueue。PriorityBlockingQueue用在需要排序的佇列中。DelayQueue可以用來做一些定時任務或者快取過期的場景。LinkedTransferQueue則相比較其他佇列多了transfer功能。最後剩下一個不儲存元素的佇列SynchronousQueue，用來處理一些高效的傳遞性場景。
參考資料：
- 《Java併發程式設計的藝術》
- 《Java併發程式設計實戰》