JUC - BlockingQueue - 天天看點

一. 前言

在新增的Concurrent包中，BlockingQueue很好的解決了多線程中，如何高效安全“傳輸”資料的問題。通過這些高效并且線程安全的隊列類，為我們快速搭建高品質的多線程程式帶來極大的便利。本文詳細介紹了BlockingQueue家庭中的所有成員，包括他們各自的功能以及常見使用場景。

二. 認識BlockingQueue

阻塞隊列，顧名思義，首先它是一個隊列，而一個隊列在資料結構中所起的作用大緻如下圖所示：

從上圖我們可以很清楚看到，通過一個共享的隊列，可以使得資料由隊列的一端輸入，從另外一端輸出；

常用的隊列主要有以下兩種：（當然通過不同的實作方式，還可以延伸出很多不同類型的隊列，DelayQueue就是其中的一種）

先進先出（FIFO）：先插入的隊列的元素也最先出隊列，類似于排隊的功能。從某種程度上來說這種隊列也展現了一種公平性。
後進先出（LIFO）：後插入隊列的元素最先出隊列，這種隊列優先處理最近發生的事件。

多線程環境中，通過隊列可以很容易實作資料共享，比如經典的“生産者”和“消費者”模型中，通過隊列可以很便利地實作兩者之間的資料共享。假設我們有若幹生産者線程，另外又有若幹個消費者線程。如果生産者線程需要把準備好的資料共享給消費者線程，利用隊列的方式來傳遞資料，就可以很友善地解決他們之間的資料共享問題。但如果生産者和消費者在某個時間段内，萬一發生資料處理速度不比對的情況呢？理想情況下，如果生産者産出資料的速度大于消費者消費的速度，并且當生産出來的資料累積到一定程度的時候，那麼生産者必須暫停等待一下（阻塞生産者線程），以便等待消費者線程把累積的資料處理完畢，反之亦然。然而，在concurrent包釋出以前，在多線程環境下，我們每個程式員都必須去自己控制這些細節，尤其還要兼顧效率和線程安全，而這會給我們的程式帶來不小的複雜度。好在此時，強大的concurrent包橫空出世了，而他也給我們帶來了強大的BlockingQueue。（在多線程領域：所謂阻塞，在某些情況下會挂起線程（即阻塞），一旦條件滿足，被挂起的線程又會自動被喚醒），下面兩幅圖示範了BlockingQueue的兩個常見阻塞場景：

如上圖所示：當隊列中沒有資料的情況下，消費者端的所有線程都會被自動阻塞（挂起），直到有資料放入隊列。

如上圖所示：當隊列中填滿資料的情況下，生産者端的所有線程都會被自動阻塞（挂起），直到隊列中有空的位置，線程被自動喚醒。

這也是我們在多線程環境下，為什麼需要BlockingQueue的原因。作為BlockingQueue的使用者，我們再也不需要關心什麼時候需要阻塞線程，什麼時候需要喚醒線程，因為這一切BlockingQueue都給你一手包辦了。既然BlockingQueue如此神通廣大，讓我們一起來見識下它的常用方法：

三. BlockingQueue的核心方法：

1. 放入資料

（1）offer(anObject):表示如果可能的話,将anObject加到BlockingQueue裡,即如果BlockingQueue可以容納,則傳回true,否則傳回false.（本方法不阻塞目前執行方法

的線程）；　　　　　　

（2）offer(E o, long timeout, TimeUnit unit)：可以設定等待的時間，如果在指定的時間内，還不能往隊列中加入BlockingQueue，則傳回失敗。

（3）put(anObject):把anObject加到BlockingQueue裡,如果BlockQueue沒有空間,則調用此方法的線程被阻斷直到BlockingQueue裡面有空間再繼續.

2. 擷取資料

（1）poll(time):取走BlockingQueue裡排在首位的對象,若不能立即取出,則可以等time參數規定的時間,取不到時傳回null;

（2）poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，如果在指定時間内，隊列一旦有資料可取，則立即傳回隊列中的資料。否則知道時間

逾時還沒有資料可取，傳回失敗。

（3）take():取走BlockingQueue裡排在首位的對象,若BlockingQueue為空,阻斷進入等待狀态直到BlockingQueue有新的資料被加入;

（4）drainTo():一次性從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：

（1）測試類

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

public class BlockingQueueTest {
 
    public static void main(String[] args) throws InterruptedException {
        // 聲明一個容量為10的緩存隊列
        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
 
        //new了三個生産者和一個消費者
        Producer producer1 = new Producer(queue);
        Producer producer2 = new Producer(queue);
        Producer producer3 = new Producer(queue);
        Consumer consumer = new Consumer(queue);
 
        // 借助Executors
        ExecutorService service = Executors.newCachedThreadPool();
        // 啟動線程
        service.execute(producer1);
        service.execute(producer2);
        service.execute(producer3);
        service.execute(consumer);
 
        // 執行10s
        Thread.sleep(10 * 1000);
        producer1.stop();
        producer2.stop();
        producer3.stop();
 
        Thread.sleep(2000);
        // 退出Executor
        service.shutdown();
    }
}

（2）生産者類

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
 
/**
 * 生産者線程
 * 
 * @author jackyuj
 */
public class Producer implements Runnable {
    
    private volatile boolean  isRunning = true;//是否在運作标志
    private BlockingQueue queue;//阻塞隊列
    private static AtomicInteger count = new AtomicInteger();//自動更新的值
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
 
    //構造函數
    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }
 
    public void run() {
        String data = null;
        Random r = new Random();
 
        System.out.println("啟動生産者線程！");
        try {
            while (isRunning) {
                System.out.println("正在生産資料...");
                Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));//取0~DEFAULT_RANGE_FOR_SLEEP值的一個随機數
 
                data = "data:" + count.incrementAndGet();//以原子方式将count目前值加1
                System.out.println("将資料：" + data + "放入隊列...");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {//設定的等待時間為2s，如果超過2s還沒加進去傳回true
                    System.out.println("放入資料失敗：" + data);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出生産者線程！");
        }
    }
 
    public void stop() {
        isRunning = false;
    }
}

（3）消費者類

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
 
/**
 * 消費者線程
 * 
 * @author jackyuj
 */
public class Consumer implements Runnable {
    
    private BlockingQueue<String> queue;
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
 
    //構造函數
    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }
 
    public void run() {
        System.out.println("啟動消費者線程！");
        Random r = new Random();
        boolean isRunning = true;
        try {
            while (isRunning) {
                System.out.println("正從隊列擷取資料...");
                String data = queue.poll(2, TimeUnit.SECONDS);//有資料時直接從隊列的隊首取走，無資料時阻塞，在2s内有資料，取走，超過2s還沒資料，傳回失敗
                if (null != data) {
                    System.out.println("拿到資料：" + data);
                    System.out.println("正在消費資料：" + data);
                    Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
                } else {
                    // 超過2s還沒資料，認為所有生産線程都已經退出，自動退出消費線程。
                    isRunning = false;
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出消費者線程！");
        }
    }
}

3. DelayQueue

DelayQueue中的元素隻有當其指定的延遲時間到了，才能夠從隊列中擷取到該元素。DelayQueue是一個沒有大小限制的隊列，是以往隊列中插入資料的操作（生産者）永遠不會被阻塞，而隻有擷取資料的操作（消費者）才會被阻塞。

使用場景：

DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個逾時未響應的連接配接隊列。

4. PriorityBlockingQueue

基于優先級的阻塞隊列（優先級的判斷通過構造函數傳入的Compator對象來決定），但需要注意的是PriorityBlockingQueue并不會阻塞資料生産者，而隻會在沒有可消費的資料時，阻塞資料的消費者。是以使用的時候要特别注意，生産者生産資料的速度絕對不能快于消費者消費資料的速度，否則時間一長，會最終耗盡所有的可用堆記憶體空間。在實作PriorityBlockingQueue時，内部控制線程同步的鎖采用的是公平鎖。

5. SynchronousQueue

一種無緩沖的等待隊列，類似于無中介的直接交易，有點像原始社會中的生産者和消費者，生産者拿着産品去集市銷售給産品的最終消費者，而消費者必須親自去集市找到所要商品的直接生産者，如果一方沒有找到合适的目标，那麼對不起，大家都在集市等待。相對于有緩沖的BlockingQueue來說，少了一個中間經銷商的環節（緩沖區），如果有經銷商，生産者直接把産品批發給經銷商，而無需在意經銷商最終會将這些産品賣給那些消費者，由于經銷商可以庫存一部分商品，是以相對于直接交易模式，總體來說采用中間經銷商的模式會吞吐量高一些（可以批量買賣）；但另一方面，又因為經銷商的引入，使得産品從生産者到消費者中間增加了額外的交易環節，單個産品的及時響應性能可能會降低。

如果采用公平模式：SynchronousQueue會采用公平鎖，并配合一個FIFO隊列來阻塞多餘的生産者和消費者，進而體系整體的公平政策；
但如果是非公平模式（SynchronousQueue預設）：SynchronousQueue采用非公平鎖，同時配合一個LIFO隊列來管理多餘的生産者和消費者，而後一種模式，如果生産者和消費者的處理速度有差距，則很容易出現饑渴的情況，即可能有某些生産者或者是消費者的資料永遠都得不到處理。

JUC - BlockingQueue

一. 前言

二. 認識BlockingQueue

三. BlockingQueue的核心方法：

四. 常見BlockingQueue

五. 小結

繼續閱讀

【JUC】公平鎖，非公平鎖源碼分析Lock

JavaScript Array對象indexOf()方法的實作

【高并發8】JUC并發容器及安全共享政策總結ArrayList -- CopyOnWriteArrayList:HashSet -- CopyOnWriteArraySetTreeSet -- ConcurrentSkipListSetHashMap -- ConcurrentHashMapTreeMap -- ConcurrentSkipListMap接下來看JUC的實際構成總結

AQS源碼解讀（二）——從acquireQueued探索獨占鎖實作原理，如何阻塞？如何喚醒？

卷妹帶你回顧Java基礎(一)每日更新Day8

數組【從入門到精通】

JAVA雜談 —— 1.1 final 修飾引用類型變量

JUC之原子類JUC原子類

Java JUC 進階線程之二原子變量和CAS算法

LeetCode-Linked List Cycle II

線程池常用的阻塞隊列有哪些？線程池常用的阻塞隊列有哪些？

了解javascript類數組

JavaScript數組reduce

ES6中數組和對象的周遊對象對象

原子整數、引用、數組、更新器-JUC-并發程式設計(Java)一、原子整數二、原子引用三、原子數組四、原子更新器

Java并發程式設計一線程池的五種狀态