Java並發編程--BlockingQueue

概述

　　BlockingQueue支持兩個附加操作的Queue：1）當Queue為空時，獲取元素線程被阻塞直到Queue變為非空；2）當Queue滿時，添加元素線程被阻塞直到Queue不滿。BlockingQueue不允許元素為null，如果入隊一個null元素，會拋NullPointerException。常用於生產者消費者模式。

　　BlockingQueue對於不能滿足條件的操作，提供了四種處理方式：

　　　　1）直接拋異常，拋出異常。如果隊列已滿，添加元素會拋出IllegalStateException異常；如果隊列為空，獲取元素會拋出NoSuchElementException異常；

　　　　2）返回一個特殊值（null或false）；

　　　　3）在滿足條件之前，無限期的阻塞當前線程，當隊列滿足條件或響應中斷退出；

　　　　4）在有限時間內阻塞當前線程，超時后返回失敗。

	拋出異常	返回特殊值	阻塞	超時
入隊	add(e)	offer(e)	put(e)	offer(e, time, unit)
出隊	remove()	poll()	take()	poll(time, unit)
檢查	element()	peek()

　　內存一致性效果：當存在其他並發 collection 時，將對象放入 BlockingQueue 之前的線程中的操作 happen-before 隨后通過另一線程從 BlockingQueue 中訪問或移除該元素的操作。

 

　　JDK提供的阻塞隊列：

　　　　ArrayBlockingQueue：一個由數組結構組成的有界阻塞隊列，遵循FIFO原則。

　　　　LinkedBlockingQueue：一個由鏈表結構組成的有界阻塞隊列，遵循FIFO原則，默認和最大長度為Integer.MAX_VALUE。

　　　　PriorityBlockingQueue：一個支持優先級排序的無界阻塞隊列。

　　　　DelayQueue：一個使用優先級隊列實現的無界阻塞隊列。

　　　　SynchronousQueue：一個不存儲元素的阻塞隊列。

　　　　LinkedTransferQueue：一個由鏈表結構組成的無界阻塞隊列。

　　　　LinkedBlockingDeque：一個由鏈表結構組成的雙向阻塞隊列。

使用

　　示例：生產者-消費者，BlockingQueue 可以安全地與多個生產者和多個使用者一起使用。

class Producer implements Runnable {
    private final BlockingQueue queue;
    Producer(BlockingQueue q) { queue = q; }
    public void run() {
        try {
            while(true) { queue.put(produce()); }    //當隊列滿時，生產者阻塞等待
        } catch (InterruptedException ex) { ... handle ...}
    }
    Object produce() { ... }
}

//消費者
class Consumer implements Runnable {
    private final BlockingQueue queue;
    Consumer(BlockingQueue q) { queue = q; }
    public void run() {
    try {
        while(true) { consume(queue.take()); }    //當隊列空時，消費者阻塞等待
    } catch (InterruptedException ex) { ... handle ...}
    }
    void consume(Object x) { ... }
}

class Setup {
    void main() {
        BlockingQueue q = new SomeQueueImplementation();
        Producer p = new Producer(q);
        Consumer c1 = new Consumer(q);
        Consumer c2 = new Consumer(q);
        new Thread(p).start();
        new Thread(c1).start();
        new Thread(c2).start();
    }
}

實現原理

　　當隊列滿時，生產者會一直阻塞，當消費者從隊列中取出元素時，如何通知生產者隊列可以繼續，以ArrayBlockingQueue和LinkedBlockingQueue為例，分析源代碼如何實現阻塞隊列。它們的阻塞機制都是基於Lock和Condition實現，其中LinkedBlockingQueue還用到了原子變量類。

　　ArrayBlockingQueue

　　　　域

/** The queued items */
final Object[] items;
/** items index for next take, poll, peek or remove */
int takeIndex;
/** items index for next put, offer, or add */
int putIndex;
/** Number of elements in the queue */
int count;
/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook.
 */

/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;

 

　　　　　　由ArrayBlockingQueue的域可以看出，使用循環數組存儲隊列中的元素，兩個索引takeIndex和putIndex分別指向下一個要出隊和入隊的數組位置，線程間的通信是使用ReentrantLock和兩個Condition實現的。

　　　　put(e)&take() 阻塞

　　　　　　當不滿足入隊或出隊條件時，當前線程阻塞等待。即當隊列滿時，生產者會一直阻塞直到被喚醒，當隊列空時，消費者會一直阻塞直到被喚醒。

　　　　　　入隊（put）

//在隊列的尾部（當前putIndex指定的位置）插入指定的元素
public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();    //可響應中斷獲取鎖
    try {
        while (count == items.length)    //如果隊列滿，在入隊條件notFull的等待隊列上等待。
                                        //這里使用While循環而非if判斷，目的是防止過早或意外的通知，只有條件符合才能推出循環
            notFull.await();
        insert(e);
    } finally {
        lock.unlock();    //釋放鎖，喚醒同步隊列中的后繼節點
    }
}
//為保證操作線程安全，此方法必須在獲取鎖的前提下才能被調用
private void insert(E x) {
    items[putIndex] = x;
    putIndex = inc(putIndex);
    ++count;            //元素數量+1
    notEmpty.signal();    //喚醒出隊條件的等待隊列上的線程
}
//將i增1，當++i等於數組的最大容量時，將i置為0。即通過循環數組的方式
final int inc(int i) {
    return (++i == items.length) ? 0 : i;
}

 

　　　　　　從源碼可以看出，入隊的大致步驟如下：

　　　　　　　　1）首先獲取鎖，如果獲取鎖失敗，當前線程可能自旋獲取鎖或被阻塞直到獲取到鎖，否則執行2）；

　　　　　　　　2）循環判斷隊列是否滿，如果滿，那么當前線程被阻塞到notFull條件的等待隊列中，並釋放鎖，等待被喚醒；

　　　　　　　　3）當隊列非滿或從await方法中返回（此時當前線程從等待隊列中被喚醒並重新獲取到鎖）時，執行插入元素操作。

　　　　　　　　4）入隊完成后，釋放鎖，喚醒同步隊列中的后繼節點。

　　　　　　出隊（take）

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();    //可響應中斷獲取鎖
    try {
        while (count == 0)    //如果隊列為空，在出隊條件notEmpty的等待隊列中等待
            notEmpty.await();
        return extract();
    } finally {
        lock.unlock();    //釋放鎖
    }
}
//在當前takeIndex指定的位置取出元素，此方法必須在獲取鎖的前提下才能被調用
private E extract() {
    final Object[] items = this.items;
    E x = this.<E>cast(items[takeIndex]);    //強制類型轉換
    items[takeIndex] = null;
    takeIndex = inc(takeIndex);    //出隊索引同樣采用循環的方式增1
    --count;
    notFull.signal();    //喚醒入隊條件的等待隊列中的線程
    return x;
}

 

　　　　　　從源碼可以看出，出隊的大致步驟如下：

　　　　　　　　1）首先獲取鎖，如果獲取鎖失敗，當前線程可能自旋獲取鎖或被阻塞直到獲取到鎖成功。

　　　　　　　　2）獲取鎖成功，循環判斷隊列是否為空，如果為空，那么當前線程被阻塞到 notEmpty 條件的等待隊列中，並釋放鎖，等待被喚醒；

　　　　　　　　3）當隊列非空或從await方法中返回（此時當前線程從等待隊列中被喚醒並重新獲取到鎖）時，執行取出元素操作。

　　　　　　　　4）出隊完成后，釋放鎖，喚醒同步隊列的后繼節點，

　　　　offer(e)&poll() 返回特殊值

　　　　　　當不能滿足入隊或出隊條件時，返回特殊值。當隊列滿時，入隊會失敗，offer方法直接返回false，反之入隊成功，返回true；當隊列空時，poll方法返回null。

　　　　　　入隊（offer）

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();    //獲取鎖
    try {
        if (count == items.length)    //如果隊列滿，與put阻塞當前線程不同的是，offer方法直接返回false
            return false;
        else {
            insert(e);
            return true;
        }
    } finally {
        lock.unlock();    //釋放鎖
    }
}

 

　　　　　　出隊（poll）

//出隊
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();    //獲取鎖
    try {
        return (count == 0) ? null : extract();    //如果隊列空，與take阻塞當前線程不同的是，poll方法返回null
    } finally {
        lock.unlock();    //釋放鎖
    }
}

 

　　　　add(e)&remove() 拋異常

　　　　　　當不能滿足入隊或出隊條件時，直接拋出異常。當隊列滿時，入隊失敗，拋IllegalStateException("Queue full")；當隊列空時，remove方法拋NoSuchElementException()異常。

　　　　　　入隊（add）

public boolean add(E e) {
    return super.add(e);
}

//抽象類AbstractQueue提供的方法
public boolean add(E e) {
    //如果offer返回true，那么add方法返回true；如果offer返回false，那么add方法拋IllegalStateException("Queue full")異常
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

　　　　　　出隊（remove）

//抽象類AbstractQueue提供的方法
public E remove() {
    E x = poll();
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

 

　　　　offer&poll 超時

　　　　　　使用Condition的超時等待機制實現，當不滿足條件時，只在有限的時間內阻塞，超過超時時間仍然不滿足條件才返回false或null。

　　　　　　入隊（offer(E e, long timeout, TimeUnit unit)）

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {

    checkNotNull(e);
    long nanos = unit.toNanos(timeout);    //轉換為納秒
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length) {
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);    //與offer直接返回false不同，此處使用Condition的超時等待機制實現，超過等待時間如果仍然不滿足條件才返回false
        }
        insert(e);
        return true;
    } finally {
        lock.unlock();
    }
}

　　　　　　出隊（poll(long timeout, TimeUnit unit)）

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);    
        }
        return extract();
    } finally {
        lock.unlock();
    }
}

 

　　LinkedBlockingQueue

　　　　Executors創建固定大小線程池的代碼，就使用了LinkedBlockingQueue來作為任務隊列。

　　　　域

/** The capacity bound, or Integer.MAX_VALUE if none */
private final int capacity;    //隊列最大容量，默認為Integer.MAX_VALUE
/** Current number of elements */
private final AtomicInteger count = new AtomicInteger(0);    //當前元素數量，原子類保證線程安全
/**
 * Head of linked list.
 * Invariant: head.item == null 
 */
private transient Node<E> head;    //隊列的首節點，head節點是個空節點，head.item == null，實際存儲元素的第一個節點是head.next
/**
 * Tail of linked list.
 * Invariant: last.next == null
 */
private transient Node<E> last;    //隊列的尾節點
/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();    //出隊鎖
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();    //出隊條件
/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();    //入隊鎖
/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();    //入隊條件

 

　　　　　　Node類：

static class Node<E> {
    E item;    //元素
    /**
     * One of:
     * - the real successor Node
     * - this Node, meaning the successor is head.next
     * - null, meaning there is no successor (this is the last node)
     */
    Node<E> next;    //后繼節點，LinkedBlockingQueue使用的是單向鏈表
    Node(E x) { item = x; }
}

 

　　　　　　由LinkedBlockingQueue的域可以看出，它使用鏈表存儲元素。線程間的通信也是使用ReentrantLock和Condition實現的，與ArrayBlockingQueue不同的是，LinkedBlockingQueue在入隊和出隊操作時分別使用兩個鎖putLock和takeLock。

　　　　　　思考問題一：為什么使用兩把鎖？

　　　　　　為了提高並發度和吞吐量，使用兩把鎖，takeLock只負責出隊，putLock只負責入隊，入隊和出隊可以同時進行，提高入隊和出隊操作的效率，增大隊列的吞吐量。LinkedBlockingQueue隊列的吞吐量通常要高於ArrayBlockingQueue隊列，但是在高並發條件下可預測性降低。

　　　　　　思考問題二：ArrayBlockingQueue中的count是一個普通的int型變量，LinkedBlockingQueue的count為什么是AtomicInteger類型的？

　　　　　　因為ArrayBlockingQueue的入隊和出隊操作使用同一把鎖，對count的修改都是在處於線程獲取鎖的情況下進行操作，因此不會有線程安全問題。而LinkedBlockingQueue的入隊和出隊操作使用的是不同的鎖，會有對count變量並發修改的情況，所以使用原子變量保證線程安全。

　　　　　　思考問題三：像notEmpty、takeLock、count域等都聲明為final型，final成員變量有什么特點？

　　　　　　1）對於一個final變量，如果是基本數據類型的變量，則其數值一旦在初始化之后便不能更改；如果是引用類型的變量，則在對其初始化之后便不能再讓其指向另一個對象。

　　　　　　2）對於一個final成員變量，必須在定義時或者構造器中進行初始化賦值，而且final變量一旦被初始化賦值之后，就不能再被賦值了。只要對象是正確構造的（被構造對象的引用在構造函數中沒有“逸出”），那么不需要使用同步（指lock和volatile的使用）就可以保證任意線程都能看到這個final域在構造函數中被初始化之后的值。

　　　　初始化

//指定容量，默認為Integer.MAX_VALUE
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);    //構造元素為null的head節點，並將last指向head節點
}

　　　　put&take 阻塞

　　　　　　阻塞式入隊（put）

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // Note: convention in all put/take/etc is to preset local var 預置本地變量，例如入隊鎖賦給局部變量putLock
    // holding count negative to indicate failure unless set.
    int c = -1;
    Node<E> node = new Node(e);    //構造新節點
    //預置本地變量putLock和count
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();    //可中斷獲取入隊鎖
    try {
        /*
         * Note that count is used in wait guard even though it is
         * not protected by lock. This works because count can
         * only decrease at this point (all other puts are shut
         * out by lock), and we (or some other waiting put) are
         * signalled if it ever changes from capacity. Similarly
         * for all other uses of count in other wait guards.
         */
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);    //在隊尾插入node
        c = count.getAndIncrement();    //count原子方式增1，返回值c為count增長之前的值
        if (c + 1 < capacity)    //如果隊列未滿，通知入隊線程（notFull條件等待隊列中的線程）
            notFull.signal();
    } finally {
        putLock.unlock();    //釋放入隊鎖
    }
    //如果入隊該元素之前隊列中元素數量為0，那么通知出隊線程（notEmpty條件等待隊列中的線程）
    if (c == 0)
        signalNotEmpty();
}

//通知出隊線程（notEmpty條件等待隊列中的線程）
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();    //獲取出隊鎖，調用notEmpty條件的方法的前提
    try {
        notEmpty.signal();    //喚醒一個等待出隊的線程
    } finally {
        takeLock.unlock();    //釋放出隊鎖
    }
}

private void enqueue(Node<E> node) {
    // assert putLock.isHeldByCurrentThread();
    // assert last.next == null;
    last = last.next = node;
}

 

　　　　　　思考問題一：為什么要再聲明一個final局部變量指向putLock和count，直接使用成員變量不行嗎？

　　　　　　直接使用成員變量：每次調用putLock的方法，都需要先通過this指針找到Heap中的Queue實例，然后在根據Queue實例的putLock域引用找到Lock實例，最后才能調用Lock的方法（即將相應的方法信息組裝成棧幀壓入棧頂）。聲明一個final局部變量指向putLock：先通過this指針找到Heap中的Queue實例，將Queue實例的putLock域存儲的Lock實例的地址賦給局部變量putLock，以后需要調用putLock的方法時，直接使用局部變量putLock引用就可以找到Lock實例。簡化了查找Lock實例的過程。count變量也是同樣的道理。個人理解應該是為了提升效率。

　　　　　　思考問題二：使用兩把鎖怎么保證元素的可見性？

　　　　　　例如：入隊線程使用put方法在隊列尾部插入一個元素，怎么保證出隊線程能看到這個元素？ArrayBlockingQueue的入隊和出隊使用同一個鎖，所以沒有可見性問題。

　　　　　　在LinkedBlockingQueue中，每次一個元素入隊， 都需要獲取putLock和更新count，而出隊線程為了保證可見性，需要獲取fullyLock（fullyLock方法用於一些批量操作,對全局加鎖）或者獲取takeLock，然后讀取count.get()。因為volatile對象的寫操作happen-before讀操作,也就是寫線程先寫的操作對隨后的讀線程是可見的，volatile相當於一個內存屏障,volatile后面的指令不允許重排序到它之前，而count是原子整型類，是基於volatile變量和CAS機制實現。所以就保證了可見性，寫線程修改count-->讀線程讀取count-->讀線程。

　　　　　　思考問題三：在put方法中，為什么喚醒出隊線程的方法signalNotEmpty()要放在釋放putLock鎖（putLock.unlock()）之后？同樣，take也有同樣的疑問？

　　　　　　避免死鎖的發生，因為signalNotEmpty()方法中要獲取takeLock鎖。如果放在釋放putLock之前，相當於在入隊線程需要先獲取putLock鎖，再獲取takeLock鎖。例如：當入隊線程先獲取到putLock鎖，並嘗試獲取takeLock鎖，出隊線程獲取到takeLock鎖，並嘗試獲取putLock鎖時，就會產生死鎖。

　　　　　　思考問題四：什么是級聯通知？

　　　　　　比如put操作會調用notEmpty的notify,只會喚醒一個等待的讀線程來take,take之后如果發現還有剩余的元素,會繼續調用notify,通知下一個線程來獲取。

　　　　　　阻塞式出隊（take）　　

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();    //可中斷獲取出隊鎖
    try {
        while (count.get() == 0) {    //如果隊列為空，阻塞線程同時釋放鎖
            notEmpty.await();
        }
        x = dequeue();    //從隊列頭彈出元素
        c = count.getAndDecrement(); //count原子式遞減
        //c>1說明本次出隊后，隊列中還有元素
        if (c > 1)
            notEmpty.signal();    //喚醒一個等待出隊的線程
    } finally {
        takeLock.unlock();    //釋放出隊鎖
    }
    //c == capacity說明本次出隊之前是滿隊列，喚醒一個等待NotFull的線程
    if (c == capacity)
        signalNotFull();
    return x;
}

//喚醒一個等待NotFull條件的線程
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

//從隊列頭彈出元素
private E dequeue() {
    // assert takeLock.isHeldByCurrentThread();
    // assert head.item == null;
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

　　　　需要注意的操作

　　　　　　1）remove

　　　　　　刪除指定元素 全局加鎖

public boolean remove(Object o) {   //正常用不到remove方法,queue正常只使用入隊出隊操作.
    if (o == null) return false;
    fullyLock();                    // 兩個鎖都鎖上了,禁止進行入隊出隊操作.
    try {
        for (Node<E> trail = head, p = trail.next;
             p != null;             // 按順序一個一個檢索,直到p==null
             trail = p, p = p.next) {
            if (o.equals(p.item)) { //找到了,就刪除掉
                unlink(p, trail);   //刪除操作是一個unlink方法,意思是p從LinkedList鏈路中解除.
                return true;        //返回刪除成功
            }
        }
        return false;
    } finally {
        fullyUnlock();
    }
}

 

　　　　　　2）contains

　　　　　　判斷是否包含指定的元素 全局加鎖

public boolean contains(Object o) {     //這種需要檢索的操作都是對全局加鎖的,很影響性能,要小心使用!
    if (o == null) return false;
    fullyLock();
    try {
        for (Node<E> p = head.next; p != null; p = p.next)
            if (o.equals(p.item))
                return true;
        return false;
    } finally {
        fullyUnlock();
    }
}

　　　　　　全局加鎖和解鎖的方法作為公共的方法供其他需要全局鎖的方法調用，避免由於獲取鎖的順序不一致導致死鎖。另外fullyLock和fullyUnlock兩個方法對鎖的操作要相反。

/**
 * Lock to prevent both puts and takes.
 */
void fullyLock() {
    putLock.lock();
    takeLock.lock();
}

/**
 * Unlock to allow both puts and takes.
 */
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

　　　　　　3）迭代器Iterator

　　　　　　弱一致性，不會不會拋出ConcurrentModificationException異常，不會阻止遍歷的時候對queue進行修改操作,可能會遍歷到修改操作的結果.

　　LinkedBlockingQueue和ArrayBlockingQueue對比

　　　　ArrayBlockingQueue由於其底層基於數組，並且在創建時指定存儲的大小，在完成后就會立即在內存分配固定大小容量的數組元素，因此其存儲通常有限，故其是一個“有界“的阻塞隊列；而LinkedBlockingQueue可以由用戶指定最大存儲容量，也可以無需指定，如果不指定則最大存儲容量將是Integer.MAX_VALUE，即可以看作是一個“無界”的阻塞隊列，由於其節點的創建都是動態創建，並且在節點出隊列后可以被GC所回收，因此其具有靈活的伸縮性。但是由於ArrayBlockingQueue的有界性，因此其能夠更好的對於性能進行預測，而LinkedBlockingQueue由於沒有限制大小，當任務非常多的時候，不停地向隊列中存儲，就有可能導致內存溢出的情況發生。

　　　　其次，ArrayBlockingQueue中在入隊列和出隊列操作過程中，使用的是同一個lock，所以即使在多核CPU的情況下，其讀取和操作的都無法做到並行，而LinkedBlockingQueue的讀取和插入操作所使用的鎖是兩個不同的lock，它們之間的操作互相不受干擾，因此兩種操作可以並行完成，故LinkedBlockingQueue的吞吐量要高於ArrayBlockingQueue。

 

 

參考資料

　　（LinkedBlockingQueue源碼分析）http://www.jianshu.com/p/cc2281b1a6bc

　　（ArrayBlockingQueue源碼分析）http://www.jianshu.com/p/9a652250e0d1

　　（源碼分析-LinkedBlockingQueue） http://blog.csdn.net/u011518120/article/details/53886256

　　（阻塞隊列LinkedBlockingQueue源碼分析）http://blog.csdn.net/levena/article/details/78322573

　　《Java並發編程的藝術》