Java併發52:併發集合系列-基於獨佔鎖+二叉樹最小堆實現的單向阻塞無界優先順序佇列PriorityBlockingQueue
原文地址：http://www.importnew.com/25541.html
一、 前言
PriorityBlockingQueue是帶優先順序的無界阻塞佇列，每次出隊都返回優先順序最高的元素，是二叉樹最小堆的實現，研究過陣列方式存放最小堆節點的都知道，直接遍歷佇列元素是無序的。
二、 PriorityBlockingQueue類圖結構
如圖PriorityBlockingQueue內部：
- 有個陣列queue用來存放佇列元素，
- size用來存放佇列元素個數，
- allocationSpinLockOffset是用來在擴容佇列時候做cas的，目的是保證只有一個執行緒可以進行擴容。
- 由於這是一個優先順序佇列所以有個比較器comparator用來比較元素大小。
- lock獨佔鎖物件用來控制同時只能有一個執行緒可以進行入隊出隊操作。
- notEmpty條件變數用來實現take方法阻塞模式。
- 這裡沒有notFull 條件變數是因為這裡的put操作是非阻塞的，為啥要設計為非阻塞的是因為這是無界佇列。
- 最後PriorityQueue q用來搞序列化的。
如下建構函式，預設佇列容量為11，預設比較器為null;
private static final int DEFAULT_INITIAL_CAPACITY = 11; public PriorityBlockingQueue() { this(DEFAULT_INITIAL_CAPACITY, null); } public PriorityBlockingQueue(int initialCapacity) { this(initialCapacity, null); } public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) { if (initialCapacity < 1) throw new IllegalArgumentException(); this.lock = new ReentrantLock(); this.notEmpty = lock.newCondition(); this.comparator = comparator; this.queue = new Object[initialCapacity]; }
三、 offer操作
在佇列插入一個元素，由於是無界佇列，所以一直為成功返回true;
public boolean offer(E e) { if (e == null) throw new NullPointerException(); final ReentrantLock lock = this.lock; lock.lock(); int n, cap; Object[] array; //如果當前元素個數>=佇列容量，則擴容(1) while ((n = size) >= (cap = (array = queue).length)) tryGrow(array, cap); try { Comparator<? super E> cmp = comparator; //預設比較器為null if (cmp == null)(2) siftUpComparable(n, e, array); else //自定義比較器(3) siftUpUsingComparator(n, e, array, cmp); //佇列元素增加1，並且啟用notEmpty的條件佇列裡面的一個阻塞執行緒 size = n + 1;（9） notEmpty.signal(); } finally { lock.unlock(); } return true; }
主流程比較簡單，下面看看兩個主要函式
private void tryGrow(Object[] array, int oldCap) { lock.unlock(); //must release and then re-acquire main lock Object[] newArray = null; //cas成功則擴容(4) if (allocationSpinLock == 0 && UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset, 0, 1)) { try { //oldGap<64則擴容新增oldcap+2,否者擴容50%，並且最大為MAX_ARRAY_SIZE int newCap = oldCap + ((oldCap < 64) ? (oldCap + 2) : // grow faster if small (oldCap >> 1)); if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow int minCap = oldCap + 1; if (minCap < 0 || minCap > MAX_ARRAY_SIZE) throw new OutOfMemoryError(); newCap = MAX_ARRAY_SIZE; } if (newCap > oldCap && queue == array) newArray = new Object[newCap]; } finally { allocationSpinLock = 0; } } //第一個執行緒cas成功後，第二個執行緒會進入這個地方，然後第二個執行緒讓出cpu，儘量讓第一個執行緒執行下面點獲取鎖，但是這得不到肯定的保證。(5) if (newArray == null) // back off if another thread is allocating Thread.yield(); lock.lock();(6) if (newArray != null && queue == array) { queue = newArray; System.arraycopy(array, 0, newArray, 0, oldCap); } }
tryGrow目的是擴容，這裡要思考下為啥在擴容前要先釋放鎖，然後使用cas控制只有一個執行緒可以擴容成功。
我的理解是為了效能，因為擴容時候是需要花時間的，如果這些操作時候還佔用鎖那麼其他執行緒在這個時候是不能進行出隊操作的，也不能進行入隊操作，這大大降低了併發性。
所以在擴容前釋放鎖，這允許其他出隊執行緒可以進行出隊操作，但是由於釋放了鎖，所以也允許在擴容時候進行入隊操作，這就會導致多個執行緒進行擴容會出現問題。
所以這裡使用了一個spinlock用cas控制只有一個執行緒可以進行擴容，失敗的執行緒呼叫Thread.yield()讓出cpu，目的意在讓擴容執行緒擴容後優先呼叫lock.lock重新獲取鎖，但是這得不到一定的保證，有可能呼叫Thread.yield()的執行緒先獲取了鎖。
那copy元素資料到新陣列為啥放到獲取鎖後面那?
原因應該是因為可見性問題，因為queue並沒有被volatile修飾。
另外有可能在擴容時候進行了出隊操作，如果直接拷貝可能看到的陣列元素不是最新的。
而通過呼叫Lock後，獲取的陣列則是最新的，並且在釋放鎖前 陣列內容不會變化。
具體建堆演算法：
private static <T> void siftUpComparable(int k, T x, Object[] array) { Comparable<? super T> key = (Comparable<? super T>) x; //佇列元素個數>0則判斷插入位置，否者直接入隊(7) while (k > 0) { int parent = (k - 1) >>> 1; Object e = array[parent]; if (key.compareTo((T) e) >= 0) break; array[k] = e; k = parent; } array[k] = key;(8) }
四、 poll操作
在佇列頭部獲取並移除一個元素，如果佇列為空，則返回null
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return dequeue(); } finally { lock.unlock(); } }
主要看dequeue
private E dequeue() { //佇列為空，則返回null int n = size - 1; if (n < 0) return null; else { //獲取隊頭元素(1) Object[] array = queue; E result = (E) array[0]; //獲取對尾元素，並值null(2) E x = (E) array[n]; array[n] = null; Comparator<? super E> cmp = comparator; if (cmp == null)//cmp=null則呼叫這個，把對尾元素位置插入到0位置，並且調整堆為最小堆(3) siftDownComparable(0, x, array, n); else siftDownUsingComparator(0, x, array, n, cmp); size = n;（4） return result; } }
private static <T> void siftDownComparable(int k, T x, Object[] array, int n) { if (n > 0) { Comparable<? super T> key = (Comparable<? super T>)x; int half = n >>> 1; // loop while a non-leaf while (k < half) { int child = (k << 1) + 1; // assume left child is least Object c = array[child];（5） int right = child + 1;（6) if (right < n && ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)(7) c = array[child = right]; if (key.compareTo((T) c) <= 0)(8) break; array[k] = c; k = child; } array[k] = key;(9) } }
五、 put操作
內部呼叫的offer,由於是無界佇列，所以不需要阻塞
public void put(E e) { offer(e); // never need to block }
六、 take操作
獲取佇列頭元素，如果佇列為空則阻塞。
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); E result; try { //如果佇列為空，則阻塞，把當前執行緒放入notEmpty的條件佇列 while ( (result = dequeue()) == null) notEmpty.await(); } finally { lock.unlock(); } return result; }
這裡是阻塞實現，阻塞後直到入隊操作呼叫notEmpty.signal 才會返回。
七、 size操作
獲取佇列元個數，由於加了獨佔鎖所以返回結果是精確的
public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return size; } finally { lock.unlock(); } }
八、總結
PriorityBlockingQueue類似於ArrayBlockingQueue內部使用一個獨佔鎖來控制同時只有一個執行緒可以進行入隊和出隊。
另外前者只使用了一個notEmpty條件變數而沒有notFull這是因為前者是無界佇列。
當put時候永遠不會處於await所以也不需要被喚醒。
PriorityBlockingQueue始終保證出隊的元素是優先順序最高的元素，並且可以定製優先順序的規則。
內部通過使用一個二叉樹最小堆演算法來維護內部陣列，這個陣列是可擴容的，當前元素個數>=最大容量時候會通過演算法擴容。
擴容：預設佇列容量為11，oldGap<64則擴容新增oldcap+2，否者擴容50%，並且最大為MAX_ARRAY_SIZE。
值得注意的是為了避免在擴容操作時候其他執行緒不能進行出隊操作，實現上使用了先釋放鎖，然後通過cas保證同時只有一個執行緒可以擴容成功。