并發(fā)容器——BlockingQueue相關(guān)類
java.util.concurrent提供了多種并發(fā)容器，總體上來說有4類
Queue類：BlockingQueue ConcurrentLinkedQueue
Map類:ConcurrentMap
Set類：ConcurrentSkipListSet CopyOnWriteArraySet
List類：CopyOnWriteArrayList
接下來一系列文章，我會對每一類的源碼進(jìn)行分析，試圖讓它們的實現(xiàn)機制完全暴露在大家面前。這篇主要是BlockingQueue及其相關(guān)類。
先給出結(jié)構(gòu)圖：
下面我按這樣的順序來展開：
1、BlockingQueue
2、ArrayBlockingQueue 2.1 添加新元素的方法：add/put/offer
2.2 該類的幾個實例變量：takeIndex/putIndex/count/
2.3 Condition實現(xiàn)
3、LinkedBlockingQueue
4、PriorityBlockingQueue
5、DelayQueue
6、BlockingDque+LinkedBlockingQueue
其中前兩個分析的盡量詳細(xì)，為了方便大家看，基本貼出了所有相關(guān)源碼。后面幾個就用盡量用文字論述，如果看得吃力，建議對著jdk的源碼看。
1、BlockingQueue
BlockingQueue繼承了Queue，Queu是先入先出(FIFO)，BlockingQueue是JDK 5.0新引入的。
根據(jù)隊列null/full時的表現(xiàn)，BlockingQueue的方法分為以下幾類：
至于為什么要使用并發(fā)容器，一個典型的例子就是生產(chǎn)者-消費者的例子，為了精簡本文篇幅，放到附件中見附件：“生產(chǎn)者-消費者 測試.rar”。
另外，BlockingQueue接口定義的所有方法實現(xiàn)都是線程安全的，它的實現(xiàn)類里面都會用鎖和其他控制并發(fā)的手段保證這種線程安全，但是這些類同時也實現(xiàn)了Collection接口(主要是AbstractQueue實現(xiàn))，所以會出現(xiàn)BlockingQueue的實現(xiàn)類也能同時使用Conllection接口方法，而這時會出現(xiàn)的問題就是像addAll,containsAll,retainAll和removeAll這類批量方法的實現(xiàn)不保證線程安全，舉個例子就是addAll 10個items到一個ArrayBlockingQueue,可能中途失敗但是卻有幾個item已經(jīng)被放進(jìn)這個隊列里面了。
2、ArrayBlockingQueue
ArrayBlockingQueue創(chuàng)建的時候需要指定容量capacity(可以存儲的最大的元素個數(shù)，因為它不會自動擴容)以及是否為公平鎖(fair參數(shù))。
在創(chuàng)建ArrayBlockingQueue的時候默認(rèn)創(chuàng)建的是非公平鎖，不過我們可以在它的構(gòu)造函數(shù)里指定。這里調(diào)用ReentrantLock的構(gòu)造函數(shù)創(chuàng)建鎖的時候，調(diào)用了：
public ReentrantLock(boolean fair) {
sync = (fair)? new FairSync() : new NonfairSync();
}
FairSync/ NonfairSync是ReentrantLock的內(nèi)部類：
線程按順序請求獲得公平鎖，而一個非公平鎖可以闖入，如果鎖的狀態(tài)可用，請求非公平鎖的線程可在等待隊列中向前跳躍，獲得該鎖。內(nèi)部鎖synchronized沒有提供確定的公平性保證。
分三點來講這個類：
2.1 添加新元素的方法：add/put/offer
2.2 該類的幾個實例變量：takeIndex/putIndex/count/
2.3 Condition實現(xiàn)
2.1 添加新元素的方法：add/put/offer
首先，談到添加元素的方法，首先得分析以下該類同步機制中用到的鎖：
Java代碼
- lock = new ReentrantLock(fair);
- notEmpty = lock.newCondition();//Condition Variable 1
- notFull = lock.newCondition();//Condition Variable 2
這三個都是該類的實例變量，只有一個鎖lock，然后lock實例化出兩個Condition，notEmpty/noFull分別用來協(xié)調(diào)多線程的讀寫操作。
Java代碼
- 1、
- public boolean offer(E e) {
- if (e == null) throw new NullPointerException();
- final ReentrantLock lock = this.lock;//每個對象對應(yīng)一個顯示的鎖
- lock.lock();//請求鎖直到獲得鎖（不可以被interrupte）
- try {
- if (count == items.length)//如果隊列已經(jīng)滿了
- return false;
- else {
- insert(e);
- return true;
- }
- } finally {
- lock.unlock();//
- }
- }
- 看insert方法：
- private void insert(E x) {
- items[putIndex] = x;
- //增加全局index的值。
- /*
- Inc方法體內(nèi)部：
- final int inc(int i) {
- return (++i == items.length)? 0 : i;
- }
- 這里可以看出ArrayBlockingQueue采用從前到后向內(nèi)部數(shù)組插入的方式插入新元素的。如果插完了，putIndex可能重新變?yōu)?（在已經(jīng)執(zhí)行了移除操作的前提下，否則在之前的判斷中隊列為滿）
- */
- putIndex = inc(putIndex);
- ++count;
- notEmpty.signal();//wake up one waiting thread
- }
Java代碼
- public void put(E e) throws InterruptedException {
- if (e == null) throw new NullPointerException();
- final E[] items = this.items;
- final ReentrantLock lock = this.lock;
- lock.lockInterruptibly();//請求鎖直到得到鎖或者變?yōu)閕nterrupted
- try {
- try {
- while (count == items.length)//如果滿了，當(dāng)前線程進(jìn)入noFull對應(yīng)的等waiting狀態(tài)
- notFull.await();
- } catch (InterruptedException ie) {
- notFull.signal(); // propagate to non-interrupted thread
- throw ie;
- }
- insert(e);
- } finally {
- lock.unlock();
- }
- }
Java代碼
- public boolean offer(E e, long timeout, TimeUnit unit)
- throws InterruptedException {
- if (e == null) throw new NullPointerException();
- long nanos = unit.toNanos(timeout);
- final ReentrantLock lock = this.lock;
- lock.lockInterruptibly();
- try {
- for (;;) {
- if (count != items.length) {
- insert(e);
- return true;
- }
- if (nanos <= 0)
- return false;
- try {
- //如果沒有被 signal/interruptes,需要等待nanos時間才返回
- nanos = notFull.awaitNanos(nanos);
- } catch (InterruptedException ie) {
- notFull.signal(); // propagate to non-interrupted thread
- throw ie;
- }
- }
- } finally {
- lock.unlock();
- }
- }
Java代碼
- public boolean add(E e) {
- return super.add(e);
- }
- 父類：
- public boolean add(E e) {
- if (offer(e))
- return true;
- else
- throw new IllegalStateException("Queue full");
- }
2.2 該類的幾個實例變量：takeIndex/putIndex/count
Java代碼
- 用三個數(shù)字來維護這個隊列中的數(shù)據(jù)變更：
- /** items index for next take, poll or remove */
- private int takeIndex;
- /** items index for next put, offer, or add. */
- private int putIndex;
- /** Number of items in the queue */
- private int count;
提取元素的三個方法take/poll/remove內(nèi)部都調(diào)用了這個方法：
Java代碼
- private E extract() {
- final E[] items = this.items;
- E x = items[takeIndex];
- items[takeIndex] = null;//移除已經(jīng)被提取出的元素
- takeIndex = inc(takeIndex);//策略和添加元素時相同
- --count;
- notFull.signal();//提醒其他在notFull這個Condition上waiting的線程可以嘗試工作了
- return x;
- }
從這個方法里可見，tabkeIndex維護一個可以提取/移除元素的索引位置，因為takeIndex是從0遞增的，所以這個類是FIFO隊列。
putIndex維護一個可以插入的元素的位置索引。
count顯然是維護隊列中已經(jīng)存在的元素總數(shù)。
2.3 Condition實現(xiàn)
Condition現(xiàn)在的實現(xiàn)只有java.util.concurrent.locks.AbstractQueueSynchoronizer內(nèi)部的ConditionObject，并且通過ReentranLock的newCondition()方法暴露出來，這是因為Condition的await()/sinal()一般在lock.lock()與lock.unlock()之間執(zhí)行，當(dāng)執(zhí)行condition.await()方法時，它會首先釋放掉本線程持有的鎖，然后自己進(jìn)入等待隊列。直到sinal()，喚醒后又會重新試圖去拿到鎖，拿到后執(zhí)行await()下的代碼，其中釋放當(dāng)前鎖和得到當(dāng)前鎖都需要ReentranLock的tryAcquire(int arg)方法來判定，并且享受ReentranLock的重進(jìn)入特性。
Java代碼
- public final void await() throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- //加一個新的condition等待節(jié)點
- Node node = addConditionWaiter();
- //釋放自己的鎖
- int savedState = fullyRelease(node);
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- //如果當(dāng)前線程 等待狀態(tài)時CONDITION，park住當(dāng)前線程，等待condition的signal來解除
- LockSupport.park(this);
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (node.nextWaiter != null)
- unlinkCancelledWaiters();
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- }
3、LinkedBlockingQueue
單向鏈表結(jié)構(gòu)的隊列。如果不指定容量默認(rèn)為Integer.MAX_VALUE。通過putLock和takeLock兩個鎖進(jìn)行同步，兩個鎖分別實例化notFull和notEmpty兩個Condtion，用來協(xié)調(diào)多線程的存取動作。其中某些方法(如remove,toArray,toString,clear等)的同步需要同時獲得這兩個鎖，并且總是先putLock.lock緊接著takeLock.lock(在同一方法fullyLock中)，這樣的順序是為了避免可能出現(xiàn)的死鎖情況(我也想不明白為什么會是這樣?)
4、PriorityBlockingQueue
看它的三個屬性，就基本能看懂這個類了：
Java代碼
- private final PriorityQueue
q; - private final ReentrantLock lock = new ReentrantLock(true);
- private final Condition notEmpty = lock.newCondition();
q說明，本類內(nèi)部數(shù)據(jù)結(jié)構(gòu)是PriorityQueue，至于PriorityQueue怎么排序看我之前一篇文章：http://jiadongkai-sina-com.iteye.com/blog/825683
lock說明本類使用一個lock來同步讀寫等操作。
notEmpty協(xié)調(diào)隊列是否有新元素提供，而隊列滿了以后會調(diào)用PriorityQueue的grow方法來擴容。
5、DelayQueue
Delayed接口繼承自Comparable
DelayQueue的設(shè)計目的間API文檔：
An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired. The head of the queue is that Delayed element whose delay expired furthest in the past. If no delay has expired there is no head and poll will returnnull. Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero. Even though unexpired elements cannot be removed using take or poll, they are otherwise treated as normal elements. For example, the size method returns the count of both expired and unexpired elements. This queue does not permit null elements.
因為DelayQueue構(gòu)造函數(shù)了里限定死不允許傳入comparator(之前的PriorityBlockingQueue中沒有限定死)，即只能在compare方法里定義優(yōu)先級的比較規(guī)則。再看上面這段英文，“The head of the queue is that Delayed element whose delay expired furthest in the past.”說明compare方法實現(xiàn)的時候要保證最先加入的元素最早結(jié)束延時。而 “Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero.”說明getDelay方法的實現(xiàn)必須保證延時到了返回的值變?yōu)?lt;=0的int。
上面這段英文中，還說明了：在poll/take的時候，隊列中元素會判定這個elment有沒有達(dá)到超時時間，如果沒有達(dá)到，poll返回null，而take進(jìn)入等待狀態(tài)。但是，除了這兩個方法，隊列中的元素會被當(dāng)做正常的元素來對待。例如，size方法返回所有元素的數(shù)量，而不管它們有沒有達(dá)到超時時間。而協(xié)調(diào)的Condition available只對take和poll是有意義的。
另外需要補充的是，在ScheduledThreadPoolExecutor中工作隊列類型是它的內(nèi)部類DelayedWorkQueue，而DelayedWorkQueue的Task容器是DelayQueue類型，而ScheduledFutureTask作為Delay的實現(xiàn)類作為Runnable的封裝后的Task類。也就是說ScheduledThreadPoolExecutor是通過DelayQueue優(yōu)先級判定規(guī)則來執(zhí)行任務(wù)的。
6、BlockingDque+LinkedBlockingQueue
BlockingDque為阻塞雙端隊列接口，實現(xiàn)類有LinkedBlockingDque。雙端隊列特別之處是它首尾都可以操作。LinkedBlockingDque不同于LinkedBlockingQueue，它只用一個lock來維護讀寫操作，并由這個lock實例化出兩個Condition notEmpty及notFull，而LinkedBlockingQueue讀和寫分別維護一個lock。
【編輯推薦】
