BlockingQueue 阻塞隊列

        阻塞隊列，顧名思義，首先它是一個隊列，而一個隊列在數據結構中所起的做用大體以下圖所示：

 

 

 

 

 

 

        從上圖咱們能夠很清楚看到，經過一個共享的隊列，可使得數據由隊列的一端輸入，從另一端輸出；
        經常使用的隊列主要有如下兩種：（固然經過不一樣的實現方式，還能夠延伸出不少不一樣類型的隊列，DelayQueue就是其中的一種）

1. 先進先出（FIFO）：先插入的隊列的元素也最早出隊列，相似於排隊的功能。從某種程度上來講這種隊列也體現了一種公平性。
2. 後進先出（LIFO）：後插入隊列的元素最早出隊列，這種隊列優先處理最近發生的事件。

         多線程環境中，經過隊列能夠很容易實現數據共享，好比經典的「生產者」和「消費者」模型中，經過隊列能夠很便利地實現二者之間的數據共享。假設咱們有若干生產者線程，另外又有若干個消費者線程。若是生產者線程須要把準備好的數據共享給消費者線程，利用隊列的方式來傳遞數據，就能夠很方便地解決他們之間的數據共享問題。但若是生產者和消費者在某個時間段內，萬一發生數據處理速度不匹配的狀況呢？理想狀況下，若是生產者產出數據的速度大於消費者消費的速度，而且當生產出來的數據累積到必定程度的時候，那麼生產者必須暫停等待一下（阻塞生產者線程），以便等待消費者線程把累積的數據處理完畢，反之亦然。然而，在concurrent包發佈之前，在多線程環境下，咱們每一個程序員都必須去本身控制這些細節，尤爲還要兼顧效率和線程安全，而這會給咱們的程序帶來不小的複雜度。好在此時，強大的concurrent包橫空出世了，而他也給咱們帶來了強大的BlockingQueue。（在多線程領域：所謂阻塞，在某些狀況下會掛起線程（即阻塞），一旦條件知足，被掛起的線程又會自動被喚醒）

BlockingQueue的核心方法：
放入數據：

1. offer(anObject):表示若是可能的話,將anObject加到BlockingQueue裏,即若是BlockingQueue能夠容納,則返回true,不然返回false.（本方法不阻塞當前執行方法的線程）
2. offer(E o, long timeout, TimeUnit unit),能夠設定等待的時間，若是在指定的時間內，還不能往隊列中加入BlockingQueue，則返回失敗。
3. put(anObject):把anObject加到BlockingQueue裏,若是BlockQueue沒有空間,則調用此方法的線程被阻斷直到BlockingQueue裏面有空間再繼續.

獲取數據：

1. poll(time):取走BlockingQueue裏排在首位的對象,若不能當即取出,則能夠等time參數規定的時間,取不到時返回null;
2. poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，若是在指定時間內，隊列一旦有數據可取，則當即返回隊列中的數據。不然知道時間超時尚未數據可取，返回失敗。
3. take():取走BlockingQueue裏排在首位的對象,若BlockingQueue爲空,阻斷進入等待狀態直到BlockingQueue有新的數據被加入; 
4. drainTo():一次性從BlockingQueue獲取全部可用的數據對象（還能夠指定獲取數據的個數）， 經過該方法，能夠提高獲取數據效率；不須要屢次分批加鎖或釋放鎖。

        BlockingQueue 具備 4 組不一樣的方法用於插入、移除以及對隊列中的元素進行檢查。若是請求的操做不能獲得當即執行的話，每一個方法的表現也不一樣。這些方法以下：

	拋異常	特定值	阻塞	超時
插入	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
移除	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
檢查	element(o)	peek(o)

四組不一樣的行爲方式解釋：

1. 拋異常：若是試圖的操做沒法當即執行，拋一個異常。
2. 特定值：若是試圖的操做沒法當即執行，返回一個特定的值(經常是 true / false)。
3. 阻塞：若是試圖的操做沒法當即執行，該方法調用將會發生阻塞，直到可以執行。
4. 超時：若是試圖的操做沒法當即執行，該方法調用將會發生阻塞，直到可以執行，但等待時間不會超過給定值。返回一個特定值以告知該操做是否成功(典型的是 true / false)。

        沒法向一個 BlockingQueue 中插入 null。若是你試圖插入 null，BlockingQueue 將會拋出一個 NullPointerException。
        能夠訪問到 BlockingQueue 中的全部元素，而不只僅是開始和結束的元素。好比說，你將一個對象放入隊列之中以等待處理，但你的應用想要將其取消掉。那麼你能夠調用諸如 remove(o) 方法來將隊列之中的特定對象進行移除。可是這麼幹效率並不高(譯者注：基於隊列的數據結構，獲取除開始或結束位置的其餘對象的效率不會過高)，所以你儘可能不要用這一類的方法，除非你確實不得不那麼作。

常見BlockingQueue
在瞭解了BlockingQueue的基本功能後，讓咱們來看看BlockingQueue家庭大體有哪些成員？ 

BlockingQueue成員詳細介紹

1. ArrayBlockingQueue
        基於數組的阻塞隊列實現，在ArrayBlockingQueue內部，維護了一個定長數組，以便緩存隊列中的數據對象，這是一個經常使用的阻塞隊列，除了一個定長數組外，ArrayBlockingQueue內部還保存着兩個整形變量，分別標識着隊列的頭部和尾部在數組中的位置。
        ArrayBlockingQueue在生產者放入數據和消費者獲取數據，都是共用同一個鎖對象，由此也意味着二者沒法真正並行運行，這點尤爲不一樣於LinkedBlockingQueue；按照實現原理來分析，ArrayBlockingQueue徹底能夠採用分離鎖，從而實現生產者和消費者操做的徹底並行運行。Doug Lea之因此沒這樣去作，也許是由於ArrayBlockingQueue的數據寫入和獲取操做已經足夠輕巧，以致於引入獨立的鎖機制，除了給代碼帶來額外的複雜性外，其在性能上徹底佔不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不一樣之處在於，前者在插入或刪除元素時不會產生或銷燬任何額外的對象實例，然後者則會生成一個額外的Node對象。這在長時間內須要高效併發地處理大批量數據的系統中，其對於GC的影響仍是存在必定的區別。而在建立ArrayBlockingQueue時，咱們還能夠控制對象的內部鎖是否採用公平鎖，默認採用非公平鎖。

2. LinkedBlockingQueue
        基於鏈表的阻塞隊列，同ArrayListBlockingQueue相似，其內部也維持着一個數據緩衝隊列（該隊列由一個鏈表構成），當生產者往隊列中放入一個數據時，隊列會從生產者手中獲取數據，並緩存在隊列內部，而生產者當即返回；只有當隊列緩衝區達到最大值緩存容量時（LinkedBlockingQueue能夠經過構造函數指定該值），纔會阻塞生產者隊列，直到消費者從隊列中消費掉一份數據，生產者線程會被喚醒，反之對於消費者這端的處理也基於一樣的原理。而LinkedBlockingQueue之因此可以高效的處理併發數據，還由於其對於生產者端和消費者端分別採用了獨立的鎖來控制數據同步，這也意味着在高併發的狀況下生產者和消費者能夠並行地操做隊列中的數據，以此來提升整個隊列的併發性能。
做爲開發者，咱們須要注意的是，若是構造一個LinkedBlockingQueue對象，而沒有指定其容量大小，LinkedBlockingQueue會默認一個相似無限大小的容量（Integer.MAX_VALUE），這樣的話，若是生產者的速度一旦大於消費者的速度，也許尚未等到隊列滿阻塞產生，系統內存就有可能已被消耗殆盡了。

ArrayBlockingQueue和LinkedBlockingQueue是兩個最普通也是最經常使用的阻塞隊列，通常狀況下，在處理多線程間的生產者消費者問題，使用這兩個類足以。

下面的代碼演示瞭如何使用BlockingQueue：

public class BlockingQueueTest {

    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) {
        // 聲明一個容量爲10的緩存隊列
        BlockingQueue sharedQueue = new LinkedBlockingQueue(10);

        //Creating Producer and Consumer Thread
        Thread prodThread = new Thread(new Producer(sharedQueue));
        Thread consThread = new Thread(new Consumer(sharedQueue));
        Thread consThread1 = new Thread(new Consumer(sharedQueue));

        //Starting producer and Consumer thread
        prodThread.start();
        consThread.start();
        consThread1.start();
    }

}

/**
 * 生產者線程
 */
public class Producer implements Runnable {
 
    private final BlockingQueue sharedQueue;
 
    public Producer(BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }
 
    @Override
    public void run() {
        for(int i=0; i<10; i++){
            try {
                System.out.println("Produced: " + i);
                sharedQueue.put(i);
            } catch (InterruptedException ex) {
                System.out.println(ex);
            }
        }
    }
 
}

/**
 * 消費者線程
 */
public class Consumer implements Runnable {

    private final BlockingQueue sharedQueue;

    public Consumer(BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                System.out.println(Thread.currentThread().getName()+"Consumed: " + sharedQueue.take());
            } catch (InterruptedException ex) {
                System.out.println(ex);
            }
        }
    }

}

output:

Produced: 0
Produced: 1
Produced: 2
Produced: 3
Produced: 4
Produced: 5
Produced: 6
Produced: 7
Produced: 8
Produced: 9
Thread-1Consumed: 0
Thread-1Consumed: 2
Thread-1Consumed: 3
Thread-1Consumed: 4
Thread-1Consumed: 5
Thread-1Consumed: 6
Thread-1Consumed: 7
Thread-1Consumed: 8
Thread-1Consumed: 9
Thread-2Consumed: 1

3. DelayQueue
        DelayQueue中的元素只有當其指定的延遲時間到了，纔可以從隊列中獲取到該元素。DelayQueue是一個沒有大小限制的隊列，所以往隊列中插入數據的操做（生產者）永遠不會被阻塞，而只有獲取數據的操做（消費者）纔會被阻塞。
使用場景：
        DelayQueue使用場景較少，但都至關巧妙，常見的例子好比使用一個DelayQueue來管理一個超時未響應的鏈接隊列。

4. PriorityBlockingQueue
        基於優先級的阻塞隊列（優先級的判斷經過構造函數傳入的Compator對象來決定），但須要注意的是PriorityBlockingQueue並不會阻塞數據生產者，而只會在沒有可消費的數據時，阻塞數據的消費者。所以使用的時候要特別注意，生產者生產數據的速度絕對不能快於消費者消費數據的速度，不然時間一長，會最終耗盡全部的可用堆內存空間。在實現PriorityBlockingQueue時，內部控制線程同步的鎖採用的是公平鎖。

5. SynchronousQueue
      一種無緩衝的等待隊列，相似於無中介的直接交易，它的內部同時只可以容納單個元素。若是該隊列已有一元素的話，試圖向隊列中插入一個新元素的線程將會阻塞，直到另外一個線程將該元素從隊列中抽走。一樣，若是該隊列爲空，試圖向隊列中抽取一個元素的線程將會阻塞，直到另外一個線程向隊列中插入了一條新的元素。

        有點像原始社會中的生產者和消費者，生產者拿着產品去集市銷售給產品的最終消費者，而消費者必須親自去集市找到所要商品的直接生產者，若是一方沒有找到合適的目標，那麼對不起，你們都在集市等待。相對於有緩衝的BlockingQueue來講，少了一箇中間經銷商的環節（緩衝區），若是有經銷商，生產者直接把產品批發給經銷商，而無需在乎經銷商最終會將這些產品賣給那些消費者，因爲經銷商能夠庫存一部分商品，所以相對於直接交易模式，整體來講採用中間經銷商的模式會吞吐量高一些（能夠批量買賣）；但另外一方面，又由於經銷商的引入，使得產品從生產者到消費者中間增長了額外的交易環節，單個產品的及時響應性能可能會下降。

        聲明一個SynchronousQueue有兩種不一樣的方式，它們之間有着不太同樣的行爲。公平模式和非公平模式的區別:
　　若是採用公平模式：SynchronousQueue會採用公平鎖，並配合一個FIFO隊列來阻塞多餘的生產者和消費者，從而體系總體的公平策略；
　　但若是是非公平模式（SynchronousQueue默認）：SynchronousQueue採用非公平鎖，同時配合一個LIFO隊列來管理多餘的生產者和消費者，然後一種模式，若是生產者和消費者的處理速度有差距，則很容易出現飢渴的狀況，便可能有某些生產者或者是消費者的數據永遠都得不處處理。

下面使用SynchronousQueue模擬只能生產一個產品的生產者-消費者模型：

public class SynchronousQueueTest {

    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Thread(new Customer(queue)).start();
        new Thread(new Product(queue)).start();
    }
}

public class Product implements Runnable {

    SynchronousQueue<Integer> queue;

    public Product(SynchronousQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            int rand = new Random().nextInt(1000);
            System.out.println("生產了一個產品：" + rand);
            System.out.println("等待三秒後運送出去...");
            try {
                TimeUnit.SECONDS.sleep(3);
                queue.put(rand);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

public class Customer implements Runnable {

    SynchronousQueue<Integer> queue;

    public Customer(SynchronousQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                System.out.println("消費了一個產品:" + queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("------------------------------------------");
        }
    }

}

output:

生產了一個產品：827
等待三秒後運送出去...
消費了一個產品:827
------------------------------------------
生產了一個產品：561
等待三秒後運送出去...
消費了一個產品:561
------------------------------------------
生產了一個產品：392
等待三秒後運送出去...
消費了一個產品:392
------------------------------------------
生產了一個產品：790
等待三秒後運送出去...
消費了一個產品:790
------------------------------------------
生產了一個產品：380

小結： 　　BlockingQueue不光實現了一個完整隊列所具備的基本功能，同時在多線程環境下，他還自動管理了多線間的自動等待於喚醒功能，從而使得程序員能夠忽略這些細節，關注更高級的功能。