Java多线程（六）之Deque与LinkedBlockingDeque深入分析
一、雙向隊列Deque
?
Queue除了前面介紹的實現外，還有一種雙向的Queue實現Deque。這種隊列允許在隊列頭和尾部進行入隊出隊操作，因此在功能上比Queue顯然要更復雜。下圖描述的是Deque的完整體系圖。需要說明的是LinkedList也已經加入了Deque的一部分（LinkedList是從jdk1.2 開始就存在數據結構）。
?
Deque在Queue的基礎上增加了更多的操作方法。
從上圖可以看到，Deque不僅具有FIFO的Queue實現，也有FILO的實現，也就是不僅可以實現隊列，也可以實現一個堆棧。
同時在Deque的體系結構圖中可以看到，實現一個Deque可以使用數組（ArrayDeque），同時也可以使用鏈表（LinkedList），還可以同實現一個支持阻塞的線程安全版本隊列LinkedBlockingDeque。
?
1、ArrayDeque實現Deque
?
?
對于數組實現的Deque來說，數據結構上比較簡單，只需要一個存儲數據的數組以及頭尾兩個索引即可。由于數組是固定長度的，所以很容易就得到數組的頭和尾，那么對于數組的操作只需要移動頭和尾的索引即可。
特別說明的是ArrayDeque并不是一個固定大小的隊列，每次隊列滿了以后就將隊列容量擴大一倍（doubleCapacity()），因此加入一個元素總是能成功，而且也不會拋出一個異常。也就是說ArrayDeque是一個沒有容量限制的隊列。
同樣繼續性能的考慮，使用System.arraycopy復制一個數組比循環設置要高效得多。
?
1.1、ArrayDeque的源碼解析
?
?
//數組雙端隊列ArrayDeque的源碼解析
public class ArrayDeque<E> extends AbstractCollection<E> implements Deque<E>, Cloneable, Serializable{
/**
* 存放隊列元素的數組，數組的長度為“2的指數”
*/
private transient E[] elements;
/**
*隊列的頭部索引位置，（被remove（）或pop（）操作的位置），當為空隊列時，首尾index相同
*/
private transient int head;
/**
* 隊列的尾部索引位置，(被 addLast(E), add(E), 或 push(E)操作的位置).
*/
private transient int tail;
/**
* 隊列的最小容量（大小必須為“2的指數”）
*/
private static final int MIN_INITIAL_CAPACITY = 8;
// ****** Array allocation and resizing utilities ******
/**
* 根據所給的數組長度，得到一個比該長度大的最小的2^p的真實長度，并建立真實長度的空數組
*/
private void allocateElements(int numElements) {
int initialCapacity = MIN_INITIAL_CAPACITY;
if (numElements >= initialCapacity) {
initialCapacity = numElements;
initialCapacity |= (initialCapacity >>> 1);
initialCapacity |= (initialCapacity >>> 2);
initialCapacity |= (initialCapacity >>> 4);
initialCapacity |= (initialCapacity >>> 8);
initialCapacity |= (initialCapacity >>> 16);
initialCapacity++;
if (initialCapacity < 0) // Too many elements, must back off
initialCapacity >>>= 1;// Good luck allocating 2 ^ 30 elements
}
elements = (E[]) new Object[initialCapacity];
}
/**
* 當隊列首尾指向同一個引用時，擴充隊列的容量為原來的兩倍，并對元素重新定位到新數組中
*/
private void doubleCapacity() {
assert head == tail;
int p = head;
int n = elements.length;
int r = n - p; // number of elements to the right of p
int newCapacity = n << 1;
if (newCapacity < 0)
throw new IllegalStateException("Sorry, deque too big");
Object[] a = new Object[newCapacity];
System.arraycopy(elements, p, a, 0, r);
System.arraycopy(elements, 0, a, r, p);
elements = (E[])a;
head = 0;
tail = n;
}
/**
* 拷貝隊列中的元素到新數組中
*/
private <T> T[] copyElements(T[] a) {
if (head < tail) {
System.arraycopy(elements, head, a, 0, size());
} else if (head > tail) {
int headPortionLen = elements.length - head;
System.arraycopy(elements, head, a, 0, headPortionLen);
System.arraycopy(elements, 0, a, headPortionLen, tail);
}
return a;
}
/**
* 默認構造隊列，初始化一個長度為16的數組
*/
public ArrayDeque() {
elements = (E[]) new Object[16];
}
/**
* 指定元素個數的構造方法
*/
public ArrayDeque(int numElements) {
allocateElements(numElements);
}
/**
* 用一個集合作為參數的構造方法
*/
public ArrayDeque(Collection<? extends E> c) {
allocateElements(c.size());
addAll(c);
}
//插入和刪除的方法主要是： addFirst(),addLast(), pollFirst(), pollLast()。
//其他的方法依賴于這些實現。
/**
* 在雙端隊列的前端插入元素,元素為null拋異常
*/
public void addFirst(E e) {
if (e == null)
throw new NullPointerException();
elements[head = (head - 1) & (elements.length - 1)] = e;
if (head == tail)
doubleCapacity();
}
/**
*在雙端隊列的末端插入元素，元素為null拋異常
*/
public void addLast(E e) {
if (e == null)
throw new NullPointerException();
elements[tail] = e;
if ( (tail = (tail + 1) & (elements.length - 1)) == head)
doubleCapacity();
}
/**
* 在前端插入，調用addFirst實現，返回boolean類型
*/
public boolean offerFirst(E e) {
addFirst(e);
return true;
}
/**
* 在末端插入，調用addLast實現，返回boolean類型
*/
public boolean offerLast(E e) {
addLast(e);
return true;
}
/**
* 刪除前端，調用pollFirst實現
*/
public E removeFirst() {
E x = pollFirst();
if (x == null)
throw new NoSuchElementException();
return x;
}
/**
* 刪除后端，調用pollLast實現
*/
public E removeLast() {
E x = pollLast();
if (x == null)
throw new NoSuchElementException();
return x;
}
//前端出對（刪除前端）
public E pollFirst() {
int h = head;
E result = elements[h]; // Element is null if deque empty
if (result == null)
return null;
elements[h] = null; // Must null out slot
head = (h + 1) & (elements.length - 1);
return result;
}
//后端出對（刪除后端）
public E pollLast() {
int t = (tail - 1) & (elements.length - 1);
E result = elements[t];
if (result == null)
return null;
elements[t] = null;
tail = t;
return result;
}
/**
* 得到前端頭元素
*/
public E getFirst() {
E x = elements[head];
if (x == null)
throw new NoSuchElementException();
return x;
}
/**
* 得到末端尾元素
*/
public E getLast() {
E x = elements[(tail - 1) & (elements.length - 1)];
if (x == null)
throw new NoSuchElementException();
return x;
}
public E peekFirst() {
return elements[head]; // elements[head] is null if deque empty
}
public E peekLast() {
return elements[(tail - 1) & (elements.length - 1)];
}
/**
* 移除此雙端隊列中第一次出現的指定元素（當從頭部到尾部遍歷雙端隊列時）。
*/
public boolean removeFirstOccurrence(Object o) {
if (o == null)
return false;
int mask = elements.length - 1;
int i = head;
E x;
while ( (x = elements[i]) != null) {
if (o.equals(x)) {
delete(i);
return true;
}
i = (i + 1) & mask;
}
return false;
}
/**
* 移除此雙端隊列中最后一次出現的指定元素（當從頭部到尾部遍歷雙端隊列時）。
*/
public boolean removeLastOccurrence(Object o) {
if (o == null)
return false;
int mask = elements.length - 1;
int i = (tail - 1) & mask;
E x;
while ( (x = elements[i]) != null) {
if (o.equals(x)) {
delete(i);
return true;
}
i = (i - 1) & mask;
}
return false;
}
// *** 隊列方法（Queue methods） ***
/**
* add方法，添加到隊列末端
*/
public boolean add(E e) {
addLast(e);
return true;
}
/**
* 同上
*/
public boolean offer(E e) {
return offerLast(e);
}
/**
* remove元素，刪除隊列前端
*/
public E remove() {
return removeFirst();
}
/**
* 彈出前端（出對，刪除前端）
*/
public E poll() {
return pollFirst();
}
public E element() {
return getFirst();
}
public E peek() {
return peekFirst();
}
// *** 棧 方法（Stack methods） ***
public void push(E e) {
addFirst(e);
}
public E pop() {
return removeFirst();
}
private void checkInvariants() { …… }
private boolean delete(int i) { …… }
// *** 集合方法（Collection Methods） ***
……
// *** Object methods ***
……
}
整體來說:1個數組，2個index（head 索引和tail索引）。實現比較簡單，容易理解。
?
?
2、LinkedList實現Deque
?
對于LinkedList本身而言，數據結構就更簡單了，除了一個size用來記錄大小外，只有head一個元素Entry。對比Map和Queue的其它數據結構可以看到這里的Entry有兩個引用，是雙向的隊列。
在示意圖中，LinkedList總是有一個“傀儡”節點，用來描述隊列“頭部”，但是并不表示頭部元素，它是一個執行null的空節點。
隊列一開始只有head一個空元素，然后從尾部加入E1(add/addLast)，head和E1之間建立雙向鏈接。然后繼續從尾部加入E2，E2就在head和E1之間建立雙向鏈接。最后從隊列的頭部加入E3(push/addFirst)，于是E3就在E1和head之間鏈接雙向鏈接。
雙向鏈表的數據結構比較簡單，操作起來也比較容易，從事從“傀儡”節點開始，“傀儡”節點的下一個元素就是隊列的頭部，前一個元素是隊列的尾部，換句話說，“傀儡”節點在頭部和尾部之間建立了一個通道，是整個隊列形成一個循環，這樣就可以從任意一個節點的任意一個方向能遍歷完整的隊列。
同樣LinkedList也是一個沒有容量限制的隊列，因此入隊列（不管是從頭部還是尾部）總能成功。
?
3、小結?
?
上面描述的ArrayDeque和LinkedList是兩種不同方式的實現，通常在遍歷和節省內存上ArrayDeque更高效（索引更快，另外不需要Entry對象），但是在隊列擴容下LinkedList更靈活，因為不需要復制原始的隊列，某些情況下可能更高效。
同樣需要注意的上述兩個實現都不是線程安全的，因此只適合在單線程環境下使用，下面章節要介紹的LinkedBlockingDeque就是線程安全的可阻塞的Deque。事實上也應該是功能最強大的Queue實現，當然了實現起來也許會復雜一點。
?
二、雙向并發阻塞隊列?LinkedBlockingDeque
?
1、LinkedBlockingDeque數據結構
?
雙向并發阻塞隊列。所謂雙向是指可以從隊列的頭和尾同時操作，并發只是線程安全的實現，阻塞允許在入隊出隊不滿足條件時掛起線程，這里說的隊列是指支持FIFO/FILO實現的鏈表。
?
首先看下LinkedBlockingDeque的數據結構。通常情況下從數據結構上就能看出這種實現的優缺點，這樣就知道如何更好的使用工具了。
從數據結構和功能需求上可以得到以下結論：
?
2、LinkedBlockingDeque源碼分析
?
?
public class LinkedBlockingDeque<E> extends AbstractQueue<E> implements BlockingDeque<E>, java.io.Serializable {
/** 包含前驅和后繼節點的雙向鏈式結構 */
static final class Node<E> {
E item;
Node<E> prev;
Node<E> next;
Node(E x, Node<E> p, Node<E> n) {
item = x;
prev = p;
next = n;
}
}
/** 頭節點 */
private transient Node<E> first;
/** 尾節點 */
private transient Node<E> last;
/** 元素個數*/
private transient int count;
/** 隊列容量 */
private final int capacity;
/** 鎖 */
private final ReentrantLock lock = new ReentrantLock();
/** notEmpty條件 */
private final Condition notEmpty = lock.newCondition();
/** notFull條件 */
private final Condition notFull = lock.newCondition();
/** 構造方法 */
public LinkedBlockingDeque() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingDeque(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
}
public LinkedBlockingDeque(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
for (E e : c)
add(e);
}
/**
* 添加元素作為新的頭節點
*/
private boolean linkFirst(E e) {
if (count >= capacity)
return false;
++count;
Node<E> f = first;
Node<E> x = new Node<E>(e, null, f);
first = x;
if (last == null)
last = x;
else
f.prev = x;
notEmpty.signal();
return true;
}
/**
* 添加尾元素
*/
private boolean linkLast(E e) {
if (count >= capacity)
return false;
++count;
Node<E> l = last;
Node<E> x = new Node<E>(e, l, null);
last = x;
if (first == null)
first = x;
else
l.next = x;
notEmpty.signal();
return true;
}
/**
* 返回并移除頭節點
*/
private E unlinkFirst() {
Node<E> f = first;
if (f == null)
return null;
Node<E> n = f.next;
first = n;
if (n == null)
last = null;
else
n.prev = null;
--count;
notFull.signal();
return f.item;
}
/**
* 返回并移除尾節點
*/
private E unlinkLast() {
Node<E> l = last;
if (l == null)
return null;
Node<E> p = l.prev;
last = p;
if (p == null)
first = null;
else
p.next = null;
--count;
notFull.signal();
return l.item;
}
/**
* 移除節點x
*/
private void unlink(Node<E> x) {
Node<E> p = x.prev;
Node<E> n = x.next;
if (p == null) {//x是頭的情況
if (n == null)
first = last = null;
else {
n.prev = null;
first = n;
}
} else if (n == null) {//x是尾的情況
p.next = null;
last = p;
} else {//x是中間的情況
p.next = n;
n.prev = p;
}
--count;
notFull.signalAll();
}
//--------------------------------- BlockingDeque 雙端阻塞隊列方法實現
public void addFirst(E e) {
if (!offerFirst(e))
throw new IllegalStateException("Deque full");
}
public void addLast(E e) {
if (!offerLast(e))
throw new IllegalStateException("Deque full");
}
public boolean offerFirst(E e) {
if (e == null) throw new NullPointerException();
lock.lock();
try {
return linkFirst(e);
} finally {
lock.unlock();
}
}
public boolean offerLast(E e) {
if (e == null) throw new NullPointerException();
lock.lock();
try {
return linkLast(e);
} finally {
lock.unlock();
}
}
public void putFirst(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
lock.lock();
try {
while (!linkFirst(e))
notFull.await();
} finally {
lock.unlock();
}
}
public void putLast(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
lock.lock();
try {
while (!linkLast(e))
notFull.await();
} finally {
lock.unlock();
}
}
public boolean offerFirst(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
for (;;) {
if (linkFirst(e))
return true;
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
} finally {
lock.unlock();
}
}
public boolean offerLast(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
for (;;) {
if (linkLast(e))
return true;
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
} finally {
lock.unlock();
}
}
public E removeFirst() {
E x = pollFirst();
if (x == null) throw new NoSuchElementException();
return x;
}
public E removeLast() {
E x = pollLast();
if (x == null) throw new NoSuchElementException();
return x;
}
public E pollFirst() {
lock.lock();
try {
return unlinkFirst();
} finally {
lock.unlock();
}
}
public E pollLast() {
lock.lock();
try {
return unlinkLast();
} finally {
lock.unlock();
}
}
public E takeFirst() throws InterruptedException {
lock.lock();
try {
E x;
while ( (x = unlinkFirst()) == null)
notEmpty.await();
return x;
} finally {
lock.unlock();
}
}
public E takeLast() throws InterruptedException {
lock.lock();
try {
E x;
while ( (x = unlinkLast()) == null)
notEmpty.await();
return x;
} finally {
lock.unlock();
}
}
public E pollFirst(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
for (;;) {
E x = unlinkFirst();
if (x != null)
return x;
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
} finally {
lock.unlock();
}
}
public E pollLast(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
lock.lockInterruptibly();
try {
for (;;) {
E x = unlinkLast();
if (x != null)
return x;
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
} finally {
lock.unlock();
}
}
public E getFirst() {
E x = peekFirst();
if (x == null) throw new NoSuchElementException();
return x;
}
public E getLast() {
E x = peekLast();
if (x == null) throw new NoSuchElementException();
return x;
}
public E peekFirst() {
lock.lock();
try {
return (first == null) ? null : first.item;
} finally {
lock.unlock();
}
}
public E peekLast() {
lock.lock();
try {
return (last == null) ? null : last.item;
} finally {
lock.unlock();
}
}
public boolean removeFirstOccurrence(Object o) {
if (o == null) return false;
lock.lock();
try {
for (Node<E> p = first; p != null; p = p.next) {
if (o.equals(p.item)) {
unlink(p);
return true;
}
}
return false;
} finally {
lock.unlock();
}
}
public boolean removeLastOccurrence(Object o) {
if (o == null) return false;
lock.lock();
try {
for (Node<E> p = last; p != null; p = p.prev) {
if (o.equals(p.item)) {
unlink(p);
return true;
}
}
return false;
} finally {
lock.unlock();
}
}
//---------------------------------- BlockingQueue阻塞隊列 方法實現
public boolean add(E e) {
addLast(e);
return true;
}
public boolean offer(E e) {
return offerLast(e);
}
public void put(E e) throws InterruptedException {
putLast(e);
}
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
return offerLast(e, timeout, unit);
}
public E remove() {
return removeFirst();
}
public E poll() {
return pollFirst();
}
public E take() throws InterruptedException {
return takeFirst();
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
return pollFirst(timeout, unit);
}
public E element() {
return getFirst();
}
public E peek() {
return peekFirst();
}
//------------------------------------------- Stack 方法實現
public void push(E e) {
addFirst(e);
}
public E pop() {
return removeFirst();
}
//------------------------------------------- Collection 方法實現
public boolean remove(Object o) {
return removeFirstOccurrence(o);
}
public int size() {
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
public boolean contains(Object o) {
if (o == null) return false;
lock.lock();
try {
for (Node<E> p = first; p != null; p = p.next)
if (o.equals(p.item))
return true;
return false;
} finally {
lock.unlock();
}
}
boolean removeNode(Node<E> e) {
lock.lock();
try {
for (Node<E> p = first; p != null; p = p.next) {
if (p == e) {
unlink(p);
return true;
}
}
return false;
} finally {
lock.unlock();
}
}
……
}
?
3、LinkedBlockingDeque的優缺點
?
有了上面的結論再來研究LinkedBlockingDeque的優缺點。
優點當然是功能足夠強大，同時由于采用一個獨占鎖，因此實現起來也比較簡單。所有對隊列的操作都加鎖就可以完成。同時獨占鎖也能夠很好的支持雙向阻塞的特性。
凡事有利必有弊。缺點就是由于獨占鎖，所以不能同時進行兩個操作，這樣性能上就大打折扣。從性能的角度講LinkedBlockingDeque要比LinkedBlockingQueue要低很多，比CocurrentLinkedQueue就低更多了，這在高并發情況下就比較明顯了。
前面分析足夠多的Queue實現后，LinkedBlockingDeque的原理和實現就不值得一提了，無非是在獨占鎖下對一個鏈表的普通操作。
?
4、LinkedBlockingDeque的序列化、反序列化
?
有趣的是此類支持序列化，但是Node并不支持序列化，因此fist/last就不能序列化，那么如何完成序列化/反序列化過程呢？
清單4 LinkedBlockingDeque的序列化、反序列化
?
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
lock.lock();
try {
// Write out capacity and any hidden stuff
s.defaultWriteObject();
// Write out all elements in the proper order.
for (Node<E> p = first; p != null; p = p.next)
s.writeObject(p.item);
// Use trailing null as sentinel
s.writeObject(null);
} finally {
lock.unlock();
}
}
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
count = 0;
first = null;
last = null;
// Read in all elements and place in queue
for (;;) {
E item = (E)s.readObject();
if (item == null)
break;
add(item);
}
}
?
清單4 描述的是LinkedBlockingDeque序列化/反序列化的過程。序列化時將真正的元素寫入輸出流，最后還寫入了一個null。讀取的時候將所有對象列表讀出來，如果讀取到一個null就表示結束。這就是為什么寫入的時候寫入一個null的原因，因為沒有將count寫入流，所以就靠null來表示結束，省一個整數空間。
?
參考內容來源：
?
集合框架 Queue篇（1）---ArrayDeque
http://hi.baidu.com/yao1111yao/item/1a1346f65a50d9c8521c266d
集合框架 Queue篇（7）---LinkedBlockingDeque
http://hi.baidu.com/yao1111yao/item/b1649cff2cf60be91a111f6d
深入淺出 Java Concurrency (24): 并發容器 part 9 雙向隊列集合 Deque
http://www.blogjava.net/xylz/archive/2010/08/12/328587.html
深入淺出 Java Concurrency (25): 并發容器 part 10 雙向并發阻塞隊列 BlockingDeque
http://www.blogjava.net/xylz/archive/2010/08/18/329227.html
總結
以上是生活随笔為你收集整理的Java多线程（六）之Deque与LinkedBlockingDeque深入分析的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Java多线程（五）之BlockingQ
- 下一篇： Java多线程（四）之Concurren