xylz,imxylz

在《并發(fā)容器 part 4 并發(fā)隊(duì)列與Queue簡介》節(jié)中的類圖中可以看到，對于Queue來說，BlockingQueue是主要的線程安全版本。這是一個(gè)可阻塞的版本，也就是允許添加/刪除元素被阻塞，直到成功為止。

BlockingQueue相對于Queue而言增加了兩個(gè)操作：put/take。下面是一張整理的表格。

看似簡單的API，非常有用。這在控制隊(duì)列的并發(fā)上非常有好處。既然加入隊(duì)列和移除隊(duì)列能夠被阻塞，這在實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模型上就簡單多了。

清單1 是生產(chǎn)者-消費(fèi)者模型的一個(gè)例子。這個(gè)例子是一個(gè)真實(shí)的場景。服務(wù)端（ICE服務(wù)）接受客戶端的請求(accept)，請求計(jì)算此人的好友生日，然后將計(jì)算的結(jié)果存取緩存中（Memcache）中。在這個(gè)例子中采用了ExecutorService實(shí)現(xiàn)多線程的功能，盡可能的提高吞吐量，這個(gè)在后面線程池的部分會(huì)詳細(xì)說明。目前就可以理解為new Thread(r).start()就可以了。另外這里阻塞隊(duì)列使用的是LinkedBlockingQueue。

清單1 一個(gè)生產(chǎn)者-消費(fèi)者例子

package xylz.study.concurrency;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class BirthdayService {

    final int workerNumber;

    final Worker[] workers;

    final ExecutorService threadPool;

    static volatile boolean running = true;

    public BirthdayService(int workerNumber, int capacity) {
        if (workerNumber <= 0) throw new IllegalArgumentException();
        this.workerNumber = workerNumber;
        workers = new Worker[workerNumber];
        for (int i = 0; i < workerNumber; i++) {
            workers[i] = new Worker(capacity);
        }
        //
        boolean b = running;// kill the resorting
        threadPool = Executors.newFixedThreadPool(workerNumber);
        for (Worker w : workers) {
            threadPool.submit(w);
        }
    }

    Worker getWorker(int id) {
        return workers[id % workerNumber];

    }

    class Worker implements Runnable {

        final BlockingQueue<Integer> queue;

        public Worker(int capacity) {
            queue = new LinkedBlockingQueue<Integer>(capacity);
        }

        public void run() {
            while (true) {
                try {
                    consume(queue.take());
                } catch (InterruptedException e) {
                    return;
                }
            }
        }

        void put(int id) {
            try {
                queue.put(id);
            } catch (InterruptedException e) {
                return;
            }
        }
    }

    public void accept(int id) {
        //accept client request
        getWorker(id).put(id);
    }

    protected void consume(int id) {
        //do the work
        //get the list of friends and save the birthday to cache
    }
}

在清單1 中可以看到不管是put()還是get()，都拋出了一個(gè)InterruptedException。我們就從這里開始，為什么會(huì)拋出這個(gè)異常。

上一節(jié)中提到實(shí)現(xiàn)一個(gè)并發(fā)隊(duì)列有三種方式。顯然只有第二種 Lock 才能實(shí)現(xiàn)阻塞隊(duì)列。在鎖機(jī)制中提到過，Lock結(jié)合Condition就可以實(shí)現(xiàn)線程的阻塞，這在鎖機(jī)制部分的很多工具中都詳細(xì)介紹過，而接下來要介紹的LinkedBlockingQueue就是采用這種方式。

LinkedBlockingQueue 原理

對比ConcurrentLinkedQueue的結(jié)構(gòu)圖，LinkedBlockingQueue多了兩個(gè)ReentrantLock和兩個(gè)Condition以及用于計(jì)數(shù)的AtomicInteger，顯然這會(huì)導(dǎo)致LinkedBlockingQueue的實(shí)現(xiàn)有點(diǎn)復(fù)雜。對照此結(jié)構(gòu)，有以下幾點(diǎn)說明：

1. 但是整體上講，LinkedBlockingQueue和ConcurrentLinkedQueue的結(jié)構(gòu)類似，都是采用頭尾節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)指向下一個(gè)節(jié)點(diǎn)的結(jié)構(gòu)，這表示它們在操作上應(yīng)該類似。
2. LinkedBlockingQueue引入了原子計(jì)數(shù)器count，這意味著獲取隊(duì)列大小size()已經(jīng)是常量時(shí)間了，不再需要遍歷隊(duì)列。每次隊(duì)列長度有變更時(shí)只需要修改count即可。
3. 有了修改Node指向有了鎖，所以不需要volatile特性了。既然有了鎖Node的item為什么需要volatile在后面會(huì)詳細(xì)分析，暫且不表。
4. 引入了兩個(gè)鎖，一個(gè)入隊(duì)列鎖，一個(gè)出隊(duì)列鎖。當(dāng)然同時(shí)有一個(gè)隊(duì)列不滿的Condition和一個(gè)隊(duì)列不空的Condition。其實(shí)參照鎖機(jī)制前面介紹過的生產(chǎn)者-消費(fèi)者模型就知道，入隊(duì)列就代表生產(chǎn)者，出隊(duì)列就代表消費(fèi)者。為什么需要兩個(gè)鎖？一個(gè)鎖行不行？其實(shí)一個(gè)鎖完全可以，但是一個(gè)鎖意味著入隊(duì)列和出隊(duì)列同時(shí)只能有一個(gè)在進(jìn)行，另一個(gè)必須等待其釋放鎖。而從ConcurrentLinkedQueue的實(shí)現(xiàn)原理來看，事實(shí)上head和last (ConcurrentLinkedQueue中是tail)是分離的，互相獨(dú)立的，這意味著入隊(duì)列實(shí)際上是不會(huì)修改出隊(duì)列的數(shù)據(jù)的，同時(shí)出隊(duì)列也不會(huì)修改入隊(duì)列，也就是說這兩個(gè)操作是互不干擾的。更通俗的將，這個(gè)鎖相當(dāng)于兩個(gè)寫入鎖，入隊(duì)列是一種寫操作，操作head，出隊(duì)列是一種寫操作，操作tail。可見它們是無關(guān)的。但是并非完全無關(guān)，后面詳細(xì)分析。

在沒有揭示入隊(duì)列和出隊(duì)列過程前，暫且猜測下實(shí)現(xiàn)原理。

根據(jù)前面學(xué)到的鎖機(jī)制原理結(jié)合ConcurrentLinkedQueue的原理，入隊(duì)列的阻塞過程大概是這樣的：

1. 獲取入隊(duì)列的鎖putLock，檢測隊(duì)列大小，如果隊(duì)列已滿，那么就掛起線程，等待隊(duì)列不滿信號notFull的喚醒。
2. 將元素加入到隊(duì)列尾部，同時(shí)修改隊(duì)列尾部引用last。
3. 隊(duì)列大小加1。
4. 釋放鎖putLock。
5. 喚醒notEmpty線程（如果有掛起的出隊(duì)列線程），告訴消費(fèi)者，已經(jīng)有了新的產(chǎn)品。

對比入隊(duì)列，出隊(duì)列的阻塞過程大概是這樣的：

1. 獲取出隊(duì)列的鎖takeLock，檢測隊(duì)列大小，如果隊(duì)列為空，那么就掛起線程，等待隊(duì)列不為空notEmpty的喚醒。
2. 將元素從頭部移除，同時(shí)修改隊(duì)列頭部引用head。
3. 隊(duì)列大小減1。
4. 釋放鎖takeLock。
5. 喚醒notFull線程（如果有掛起的入隊(duì)列線程），告訴生產(chǎn)者，現(xiàn)在還有空閑的空間。

下面來驗(yàn)證上面的過程。

入隊(duì)列過程（put/offer）

清單2 阻塞的入隊(duì)列過程

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        try {
            while (count.get() == capacity)
                notFull.await();
        } catch (InterruptedException ie) {
            notFull.signal(); // propagate to a non-interrupted thread
            throw ie;
        }
        insert(e);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

清單2 描述的是入隊(duì)列的阻塞過程。可以看到和上面描述的入隊(duì)列的過程基本相同。但是也有以下幾個(gè)問題：

1. 如果在入隊(duì)列的時(shí)候線程被中斷，那么就需要發(fā)出一個(gè)notFull的信號，表示下一個(gè)入隊(duì)列的線程能夠被喚醒（如果阻塞的話）。
2. 入隊(duì)列成功后如果隊(duì)列不滿需要補(bǔ)一個(gè)notFull的信號。為什么？隊(duì)列不滿的時(shí)候其它入隊(duì)列的阻塞線程難道不知道么？有可能。這是因?yàn)闉榱藴p少上下文切換的次數(shù)，每次喚醒一個(gè)線程（不管是入隊(duì)列還是出隊(duì)列）都是只隨機(jī)喚醒一個(gè)(notify)，而不是喚醒所有的（notifyall()）。這會(huì)導(dǎo)致其它阻塞的入隊(duì)列線程不能夠即使處理隊(duì)列不滿的情況。
3. 如果隊(duì)列不為空并且可能有一個(gè)元素的話就喚醒一個(gè)出隊(duì)列線程。這么做說明之前隊(duì)列一定為空，因?yàn)樵诩尤腙?duì)列之后隊(duì)列最多只能為1，那么說明未加入之前是0，那么就可能有被阻塞的出隊(duì)列線程，所以就喚醒一個(gè)出隊(duì)列線程。特別說明的是為什么使用一個(gè)臨時(shí)變量c，而不用count。這是因?yàn)樽x取一個(gè)count的開銷比讀取一個(gè)臨時(shí)一個(gè)變量大，而此處c又能夠完成確認(rèn)隊(duì)列最多只有一個(gè)元素的判斷。首先c默認(rèn)為-1，如果加入隊(duì)列后獲取原子計(jì)數(shù)器的結(jié)果為0，說明之前隊(duì)列為空，不可能消費(fèi)（出隊(duì)列），也不可能入隊(duì)列，因?yàn)榇藭r(shí)鎖還在當(dāng)前線程上，那么加入一個(gè)后隊(duì)列就不為空了，所以就可以安全的喚醒一個(gè)消費(fèi)（出對立）線程。
4. 入隊(duì)列的過程允許被中斷，所以總是拋出InterruptedException 異常。

針對第2點(diǎn)，特別補(bǔ)充說明下。本來這屬于鎖機(jī)制中條件隊(duì)列的范圍，由于沒有應(yīng)用場景，所以當(dāng)時(shí)沒有提。

前面提高notifyall總是比notify更可靠，因?yàn)閚otify可能丟失通知，為什么不適用notifyall呢？

先解釋下notify丟失通知的問題。

notify丟失通知問題

假設(shè)線程A因?yàn)槟撤N條件在條件隊(duì)列中等待，同時(shí)線程B因?yàn)榱硗庖环N條件在同一個(gè)條件隊(duì)列中等待，也就是說線程A/B都被同一個(gè)Conditon.await()掛起，但是等待的條件不同。現(xiàn)在假設(shè)線程B的線程被滿足，線程C執(zhí)行一個(gè)notify操作，此時(shí)JVM從Conditon.await()的多個(gè)線程（A/B）中隨機(jī)挑選一個(gè)喚醒，不幸的是喚醒了A。此時(shí)A的條件不滿足，于是A繼續(xù)掛起。而此時(shí)B仍然在傻傻的等待被喚醒的信號。也就是說本來給B的通知卻被一個(gè)無關(guān)的線程持有了，真正需要通知的線程B卻沒有得到通知，而B仍然在等待一個(gè)已經(jīng)發(fā)生過的通知。

如果使用notifyall，則能夠避免此問題。notifyall會(huì)喚醒所有正在等待的線程，線程C發(fā)出的通知線程A同樣能夠收到，但是由于對于A沒用，所以A繼續(xù)掛起，而線程B也收到了此通知，于是線程B正常被喚醒。

既然notifyall能夠解決單一notify丟失通知的問題，那么為什么不總是使用notifyall替換notify呢？

假設(shè)有N個(gè)線程在條件隊(duì)列中等待，調(diào)用notifyall會(huì)喚醒所有線程，然后這N個(gè)線程競爭同一個(gè)鎖，最多只有一個(gè)線程能夠得到鎖，于是其它線程又回到掛起狀態(tài)。這意味每一次喚醒操作可能帶來大量的上下文切換（如果N比較大的話），同時(shí)有大量的競爭鎖的請求。這對于頻繁的喚醒操作而言性能上可能是一種災(zāi)難。

如果說總是只有一個(gè)線程被喚醒后能夠拿到鎖，那么為什么不使用notify呢？所以某些情況下使用notify的性能是要高于notifyall的。

如果滿足下面的條件，可以使用單一的notify取代notifyall操作：

相同的等待者，也就是說等待條件變量的線程操作相同，每一個(gè)從wait放回后執(zhí)行相同的邏輯，同時(shí)一個(gè)條件變量的通知至多只能喚醒一個(gè)線程。

也就是說理論上講在put/take中如果使用sinallAll喚醒的話，那么在清單2 中的notFull.singal就是多余的。

出隊(duì)列過程（poll/take）

再來看出隊(duì)列過程。清單3 描述了出隊(duì)列的過程。可以看到這和入隊(duì)列是對稱的。從這里可以看到，出隊(duì)列使用的是和入隊(duì)列不同的鎖，所以入隊(duì)列、出隊(duì)列這兩個(gè)操作才能并行進(jìn)行。

清單3 阻塞的出隊(duì)列過程

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        try {
            while (count.get() == 0)
                notEmpty.await();
        } catch (InterruptedException ie) {
            notEmpty.signal(); // propagate to a non-interrupted thread
            throw ie;
        }

        x = extract();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}

為什么有異常？

有了入隊(duì)列、出隊(duì)列的過程后再來回答前面的幾個(gè)問題。

為什么總是拋出InterruptedException 異常？ 這是很大一塊內(nèi)容，其實(shí)是Java對線程中斷的處理問題，希望能夠在系列文章的最后能夠?qū)Υ碎_辟單獨(dú)的篇章來談?wù)劇?/p>

在鎖機(jī)制里面也是總遇到，這是因?yàn)椋琂ava里面沒有一種直接的方法中斷一個(gè)掛起的線程，所以通常情況下等于一個(gè)處于WAITING狀態(tài)的線程，允許設(shè)置一個(gè)中斷位，一旦線程檢測到這個(gè)中斷位就會(huì)從WAITING狀態(tài)退出，以一個(gè)InterruptedException 的異常返回。所以只要是對一個(gè)線程掛起操作都會(huì)導(dǎo)致InterruptedException 的可能，比如Thread.sleep()、Thread.join()、Object.wait()。盡管LockSupport.park()不會(huì)拋出一個(gè)InterruptedException 異常，但是它會(huì)將當(dāng)前線程的的interrupted狀態(tài)位置上，而對于Lock/Condition而言，當(dāng)捕捉到interrupted狀態(tài)后就認(rèn)為線程應(yīng)該終止任務(wù)，所以就拋出了一個(gè)InterruptedException 異常。

又見volatile

還有一個(gè)不容易理解的問題。為什么Node.item是volatile類型的？

起初我不大明白，因?yàn)閷τ谝粋€(gè)進(jìn)入隊(duì)列的Node，它的item是不變，當(dāng)且僅當(dāng)出隊(duì)列的時(shí)候會(huì)將頭結(jié)點(diǎn)元素的item 設(shè)置為null。盡管在remove(o)的時(shí)候也是設(shè)置為null,但是那時(shí)候是加了putLock/takeLock兩個(gè)鎖的，所以肯定是沒有問題的。那么問題出在哪？

我們知道，item的值是在put/offer的時(shí)候加入的。這時(shí)候都是有putLock鎖保證的，也就是說它保證使用putLock鎖的讀取肯定是沒有問題的。那么問題就只可能出在一個(gè)不適用putLock卻需要讀取Node.item的地方。

peek操作時(shí)獲取頭結(jié)點(diǎn)的元素而不移除它。顯然他不會(huì)操作尾節(jié)點(diǎn)，所以它不需要putLock鎖，也就是說它只有takeLock鎖。清單4 描述了這個(gè)過程。

清單4 查詢隊(duì)列頭元素過程

public E peek() {
    if (count.get() == 0)
        return null;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        Node<E> first = head.next;
        if (first == null)
            return null;
        else
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

清單4 描述了peek的過程，最后返回一個(gè)非null節(jié)點(diǎn)的結(jié)果是Node.item。這里讀取了Node的item值，但是整個(gè)過程卻是使用了takeLock而非putLock。換句話說putLock對Node.item的操作，peek()線程可能不可見！

清單5 隊(duì)列尾部加入元素

private void insert(E x) {
    last = last.next = new Node<E>(x);
}

 

清單5 是入隊(duì)列offer/put的一部分，這里關(guān)鍵在于last=new Node<E>(x)可能發(fā)生重排序。Node構(gòu)造函數(shù)是這樣的：Node(E x) { item = x; }。在這一步里面我們可能得到以下一種情況：

1. 構(gòu)建一個(gè)Node對象n；
2. 將Node的n賦給last
3. 初始化n，設(shè)置item=x

在執(zhí)行步驟2 的時(shí)候一個(gè)peek線程可能拿到了新的Node n，這時(shí)候它讀取item，得到了一個(gè)null。顯然這是不可靠的。

對item采用volatile之后，JMM保證對item=x的賦值一定在last=n之前，也就是說last得到的一個(gè)是一個(gè)已經(jīng)賦值了的新節(jié)點(diǎn)n。這就不會(huì)導(dǎo)致讀取空元素的問題的。

出對了poll/take和peek都是使用的takeLock鎖，所以不會(huì)導(dǎo)致此問題。

刪除操作和遍歷操作由于同時(shí)獲取了takeLock和putLock，所以也不會(huì)導(dǎo)致此問題。

總結(jié)：當(dāng)前僅當(dāng)元素加入隊(duì)列時(shí)讀取此元素才可能導(dǎo)致不一致的問題。采用volatile正式避免此問題。

附加功能

BlockingQueue有一個(gè)額外的功能，允許批量從隊(duì)列中異常元素。這個(gè)API是：

int drainTo(Collection<? super E> c, int maxElements); 最多從此隊(duì)列中移除給定數(shù)量的可用元素，并將這些元素添加到給定 collection 中。

int drainTo(Collection<? super E> c); 移除此隊(duì)列中所有可用的元素，并將它們添加到給定 collection 中。

清單6 描述的是最多移除指定數(shù)量元素的過程。由于批量操作只需要一次獲取鎖，所以效率會(huì)比每次獲取鎖要高。但是需要說明的，需要同時(shí)獲取takeLock/putLock兩把鎖，因?yàn)楫?dāng)移除完所有元素后這會(huì)涉及到尾節(jié)點(diǎn)的修改（last節(jié)點(diǎn)仍然指向一個(gè)已經(jīng)移走的節(jié)點(diǎn)）。

由于迭代操作contains()/remove()/iterator()也是獲取了兩個(gè)鎖，所以迭代操作也是線程安全的。

清單6 批量移除操作

public int drainTo(Collection<? super E> c, int maxElements) {
    if (c == null)
        throw new NullPointerException();
    if (c == this)
        throw new IllegalArgumentException();
    fullyLock();
    try {
        int n = 0;
        Node<E> p = head.next;
        while (p != null && n < maxElements) {
            c.add(p.item);
            p.item = null;
            p = p.next;
            ++n;
        }
        if (n != 0) {
            head.next = p;
            assert head.item == null;
            if (p == null)
                last = head;
            if (count.getAndAdd(-n) == capacity)
                notFull.signalAll();
        }
        return n;
    } finally {
        fullyUnlock();
    }
}