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