程式師世界 >> 編程語言 >> JAVA編程 >> 關於JAVA >> 並發容器——BlockingQueue相關類

並發容器——BlockingQueue相關類

編輯：關於JAVA

Java.util.concurrent提供了多種並發容器，總體上來說有4類

Queue類：BlockingQueue ConcurrentLinkedQueue

Map類:ConcurrentMap

Set類：ConcurrentSkipListSet CopyOnWriteArraySet

List類：CopyOnWriteArrayList

接下來一系列文章，我會對每一類的源碼進行分析，試圖讓它們的實現機制完全暴露在大家面前。這篇主要是BlockingQueue及其相關類。

先給出結構圖：

下面我按這樣的順序來展開：

1、BlockingQueue

2、ArrayBlockingQueue 2.1 添加新元素的方法：add/put/offer

2.2 該類的幾個實例變量：takeIndex/putIndex/count/

2.3 Condition實現

3、LinkedBlockingQueue

4、PriorityBlockingQueue

5、DelayQueue

6、BlockingDque+LinkedBlockingQueue

其中前兩個分析的盡量詳細，為了方便大家看，基本貼出了所有相關源碼。後面幾個就用盡量用文字論述，如果看得吃力，建議對著jdk的源碼看。

1、BlockingQueue

BlockingQueue繼承了Queue，Queu是先入先出(FIFO)，BlockingQueue是JDK 5.0新引入的。

根據隊列null/full時的表現，BlockingQueue的方法分為以下幾類：

至於為什麼要使用並發容器，一個典型的例子就是生產者-消費者的例子，為了精簡本文篇幅，放到附件中見附件：“生產者-消費者測試.rar”。

另外，BlockingQueue接口定義的所有方法實現都是線程安全的，它的實現類裡面都會用鎖和其他控制並發的手段保證這種線程安全，但是這些類同時也實現了Collection接口(主要是AbstractQueue實現)，所以會出現BlockingQueue的實現類也能同時使用Conllection接口方法，而這時會出現的問題就是像addAll,containsAll,retainAll和removeAll這類批量方法的實現不保證線程安全，舉個例子就是addAll 10個items到一個ArrayBlockingQueue,可能中途失敗但是卻有幾個item已經被放進這個隊列裡面了。

2、ArrayBlockingQueue

ArrayBlockingQueue創建的時候需要指定容量capacity(可以存儲的最大的元素個數，因為它不會自動擴容)以及是否為公平鎖(fair參數)。

在創建ArrayBlockingQueue的時候默認創建的是非公平鎖，不過我們可以在它的構造函數裡指定。這裡調用ReentrantLock的構造函數創建鎖的時候，調用了：

public ReentrantLock(boolean fair) {

sync = (fair)? new FairSync() : new NonfairSync();

}

FairSync/ NonfairSync是ReentrantLock的內部類：

線程按順序請求獲得公平鎖，而一個非公平鎖可以闖入，如果鎖的狀態可用，請求非公平鎖的線程可在等待隊列中向前跳躍，獲得該鎖。內部鎖synchronized沒有提供確定的公平性保證。

分三點來講這個類：

2.1 添加新元素的方法：add/put/offer

2.2 該類的幾個實例變量：takeIndex/putIndex/count/

2.3 Condition實現

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分別用來協調多線程的讀寫操作。

Java代碼

1、     
public boolean offer(E e) {     
        if (e == null) throw new NullPointerException();     
        final ReentrantLock lock = this.lock;//每個對象對應一個顯示的鎖     
        lock.lock();//請求鎖直到獲得鎖（不可以被interrupte）     
        try {     
            if (count == items.length)//如果隊列已經滿了     
                return false;     
            else {     
                insert(e);     
                return true;     
            }     
        } finally {     
            lock.unlock();//     
        }     
}     
看insert方法：     
private void insert(E x) {     
        items[putIndex] = x;     
        //增加全局index的值。     
        /*    
        Inc方法體內部：    
        final int inc(int i) {    
        return (++i == items.length)? 0 : i;    
            }    
        這裡可以看出ArrayBlockingQueue采用從前到後向內部數組插入的方式插入新元素的。如果插完了，putIndex可能重新變為0（在已經執行了移除操作的前提下，否則在之前的判斷中隊列為滿）    
        */    
        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();//請求鎖直到得到鎖或者變為interrupted     
        try {     
            try {     
                while (count == items.length)//如果滿了，當前線程進入noFull對應的等waiting狀態     
                    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代碼

用三個數字來維護這個隊列中的數據變更：     
/** 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內部都調用了這個方法：

Java代碼

private E extract() {     
        final E[] items = this.items;     
        E x = items[takeIndex];     
        items[takeIndex] = null;//移除已經被提取出的元素     
        takeIndex = inc(takeIndex);//策略和添加元素時相同     
        --count;     
        notFull.signal();//提醒其他在notFull這個Condition上waiting的線程可以嘗試工作了     
        return x;     
    }

從這個方法裡可見，tabkeIndex維護一個可以提取/移除元素的索引位置，因為takeIndex是從0遞增的，所以這個類是FIFO隊列。

putIndex維護一個可以插入的元素的位置索引。

count顯然是維護隊列中已經存在的元素總數。

2.3 Condition實現

Condition現在的實現只有Java.util.concurrent.locks.AbstractQueueSynchoronizer內部的ConditionObject，並且通過ReentranLock的newCondition()方法暴露出來，這是因為Condition的await()/sinal()一般在lock.lock()與lock.unlock()之間執行，當執行condition.await()方法時，它會首先釋放掉本線程持有的鎖，然後自己進入等待隊列。直到sinal()，喚醒後又會重新試圖去拿到鎖，拿到後執行await()下的代碼，其中釋放當前鎖和得到當前鎖都需要ReentranLock的tryAcquire(int arg)方法來判定，並且享受ReentranLock的重進入特性。

Java代碼

public final void await() throws InterruptedException {     
            if (Thread.interrupted())     
                throw new InterruptedException();     
           //加一個新的condition等待節點     
 Node node = addConditionWaiter();     
//釋放自己的鎖     
            int savedState = fullyRelease(node);      
            int interruptMode = 0;     
            while (!isOnSyncQueue(node)) {     
            //如果當前線程 等待狀態時CONDITION，park住當前線程，等待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

單向鏈表結構的隊列。如果不指定容量默認為Integer.MAX_VALUE。通過putLock和takeLock兩個鎖進行同步，兩個鎖分別實例化notFull和notEmpty兩個Condtion，用來協調多線程的存取動作。其中某些方法(如remove,toArray,toString,clear等)的同步需要同時獲得這兩個鎖，並且總是先putLock.lock緊接著takeLock.lock(在同一方法fullyLock中)，這樣的順序是為了避免可能出現的死鎖情況(我也想不明白為什麼會是這樣?)

4、PriorityBlockingQueue

看它的三個屬性，就基本能看懂這個類了：

Java代碼

private final PriorityQueue q;     
    private final ReentrantLock lock = new ReentrantLock(true);     
    private final Condition notEmpty = lock.newCondition();

q說明，本類內部數據結構是PriorityQueue，至於PriorityQueue怎麼排序看我之前一篇文章：http://jiadongkai-sina-com.iteye.com/blog/825683

lock說明本類使用一個lock來同步讀寫等操作。

notEmpty協調隊列是否有新元素提供，而隊列滿了以後會調用PriorityQueue的grow方法來擴容。

5、DelayQueue

Delayed接口繼承自Comparable,我們插入的E元素都要實現這個接口。

DelayQueue的設計目的間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構造函數了裡限定死不允許傳入comparator(之前的PriorityBlockingQueue中沒有限定死)，即只能在compare方法裡定義優先級的比較規則。再看上面這段英文，“The head of the queue is that Delayed element whose delay expired furthest in the past.”說明compare方法實現的時候要保證最先加入的元素最早結束延時。而 “Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero.”說明getDelay方法的實現必須保證延時到了返回的值變為<=0的int。

上面這段英文中，還說明了：在poll/take的時候，隊列中元素會判定這個elment有沒有達到超時時間，如果沒有達到，poll返回null，而take進入等待狀態。但是，除了這兩個方法，隊列中的元素會被當做正常的元素來對待。例如，size方法返回所有元素的數量，而不管它們有沒有達到超時時間。而協調的Condition available只對take和poll是有意義的。

另外需要補充的是，在ScheduledThreadPoolExecutor中工作隊列類型是它的內部類DelayedWorkQueue，而DelayedWorkQueue的Task容器是DelayQueue類型，而ScheduledFutureTask作為Delay的實現類作為Runnable的封裝後的Task類。也就是說ScheduledThreadPoolExecutor是通過DelayQueue優先級判定規則來執行任務的。

6、BlockingDque+LinkedBlockingQueue

BlockingDque為阻塞雙端隊列接口，實現類有LinkedBlockingDque。雙端隊列特別之處是它首尾都可以操作。LinkedBlockingDque不同於LinkedBlockingQueue，它只用一個lock來維護讀寫操作，並由這個lock實例化出兩個Condition notEmpty及notFull，而LinkedBlockingQueue讀和寫分別維護一個lock。