前言

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

BlockingQueue: j.u.c包下的提供了線程安全的隊列訪問的接口，并發(fā)包下很多高級同步類的實現(xiàn)都是基于阻塞隊列列實現(xiàn)的

1、當阻塞列進行插入數(shù)據(jù)時，如果隊列已滿，線程將會阻塞等待直到隊列非滿

2、從阻塞隊列讀數(shù)據(jù)時，如果隊列為空，線程將會阻塞等待直到隊列里面是非空的時候

• ArrayBlockingQueue： 基于數(shù)組實現(xiàn)的?個阻塞隊列，需要指定容量??， FIFO先進先出順序
• LinkedBlockingQueue： 基于鏈表實現(xiàn)的?個阻塞隊列，如果不指定容量??，默認Integer.MAX_VALUE, FIFO先進先出順序
• PriorityBlockingQueue： ?個?持優(yōu)先級的?界阻塞隊列，默認情況下元素采??然順序升序排序，也可以?定義排序實現(xiàn) java.lang.Comparable接?
• DelayQueue： 延遲隊列，在指定時間才能獲取隊列元素的功能，隊列頭元素是最接近過期的元素， ??的對象必須實現(xiàn) java.util.concurrent.Delayed 接?并實現(xiàn)CompareTo和getDelay?法

認識BlockingQueue

阻塞隊列，顧名思義，首先它是一個隊列，而一個隊列在數(shù)據(jù)結構中所起的作用大致如下圖所示：

從上圖我們可以很清楚看到，通過一個共享的隊列，可以使得數(shù)據(jù)由隊列的一端輸入，從另外一端輸出；

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

先進先出（FIFO）：先插入的隊列的元素也最先出隊列，類似于排隊的功能。從某種程度上來說這種隊列也體現(xiàn)了一種公平性。

后進先出（LIFO）：后插入隊列的元素最先出隊列，這種隊列優(yōu)先處理最近發(fā)生的事件。

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

如上圖所示：當隊列中沒有數(shù)據(jù)的情況下，消費者端的所有線程都會被自動阻塞（掛起），直到有數(shù)據(jù)放入隊列。

如上圖所示：當隊列中填滿數(shù)據(jù)的情況下，生產者端的所有線程都會被自動阻塞（掛起），直到隊列中有空的位置，線程被自動喚醒。 　

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

BlockingQueue的核心方法

放入數(shù)據(jù)

（1）offer(anObject):表示如果可能的話,將anObject加到BlockingQueue里,即如果BlockingQueue可以容納,則返回true,否則返回false.（本方法不阻塞當前執(zhí)行方法的線程）；

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

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

獲取數(shù)據(jù)

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

（2）poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，如果在指定時間內，隊列一旦有數(shù)據(jù)可取，則立即返回隊列中的數(shù)據(jù)。否則知道時間超時還沒有數(shù)據(jù)可取，返回失敗。

（3）take():取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻斷進入等待狀態(tài)直到BlockingQueue有新的數(shù)據(jù)被加入;

（4）drainTo():一次性從BlockingQueue獲取所有可用的數(shù)據(jù)對象（還可以指定獲取數(shù)據(jù)的個數(shù)），通過該方法，可以提升獲取數(shù)據(jù)效率；不需要多次分批加鎖或釋放鎖。

常見BlockingQueue

在了解了BlockingQueue的基本功能后，讓我們來看看BlockingQueue家庭大致有哪些成員？

ArrayBlockingQueue

基于數(shù)組的阻塞隊列實現(xiàn)，在ArrayBlockingQueue內部，維護了一個定長數(shù)組，以便緩存隊列中的數(shù)據(jù)對象，這是一個常用的阻塞隊列，除了一個定長數(shù)組外，ArrayBlockingQueue內部還保存著兩個整形變量，分別標識著隊列的頭部和尾部在數(shù)組中的位置。

ArrayBlockingQueue在生產者放入數(shù)據(jù)和消費者獲取數(shù)據(jù)，都是共用同一個鎖對象，由此也意味著兩者無法真正并行運行，這點尤其不同于LinkedBlockingQueue；按照實現(xiàn)原理來分析，ArrayBlockingQueue完全可以采用分離鎖，從而實現(xiàn)生產者和消費者操作的完全并行運行。Doug Lea之所以沒這樣去做，也許是因為ArrayBlockingQueue的數(shù)據(jù)寫入和獲取操作已經足夠輕巧，以至于引入獨立的鎖機制，除了給代碼帶來額外的復雜性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在于，前者在插入或刪除元素時不會產生或銷毀任何額外的對象實例，而后者則會生成一個額外的Node對象。這在長時間內需要高效并發(fā)地處理大批量數(shù)據(jù)的系統(tǒng)中，其對于GC的影響還是存在一定的區(qū)別。而在創(chuàng)建ArrayBlockingQueue時，我們還可以控制對象的內部鎖是否采用公平鎖，默認采用非公平鎖。

LinkedBlockingQueue

基于鏈表的阻塞隊列，同ArrayListBlockingQueue類似，其內部也維持著一個數(shù)據(jù)緩沖隊列（該隊列由一個鏈表構成），當生產者往隊列中放入一個數(shù)據(jù)時，隊列會從生產者手中獲取數(shù)據(jù)，并緩存在隊列內部，而生產者立即返回；只有當隊列緩沖區(qū)達到最大值緩存容量時（LinkedBlockingQueue可以通過構造函數(shù)指定該值），才會阻塞生產者隊列，直到消費者從隊列中消費掉一份數(shù)據(jù)，生產者線程會被喚醒，反之對于消費者這端的處理也基于同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理并發(fā)數(shù)據(jù)，還因為其對于生產者端和消費者端分別采用了獨立的鎖來控制數(shù)據(jù)同步，這也意味著在高并發(fā)的情況下生產者和消費者可以并行地操作隊列中的數(shù)據(jù)，以此來提高整個隊列的并發(fā)性能。

作為開發(fā)者，我們需要注意的是，如果構造一個LinkedBlockingQueue對象，而沒有指定其容量大小，LinkedBlockingQueue會默認一個類似無限大小的容量（Integer.MAX_VALUE），這樣的話，如果生產者的速度一旦大于消費者的速度，也許還沒有等到隊列滿阻塞產生，系統(tǒng)內存就有可能已被消耗殆盡了。

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

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

生產者線程

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: ruan
 * Date: 2021/9/25 17:36
 * @Description: 生產者線程
 */
public class Producer implements Runnable {
    /**
     * 標識位---是否運行中
     */
    private volatile boolean isRunning = true;
    /**
     * 創(chuàng)建阻塞隊列
     */
    private BlockingQueue<String> queue;
    /**
     * 用于記錄庫存量
     */
    private static AtomicInteger count = new AtomicInteger(0);

    public Producer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    /**
     * 構造函數(shù)
     */


    @Override
    public void run() {
        String data = null;
        Random random = new Random();
        System.out.println("生產者線程啟動....");
        while (isRunning){
            try {
                System.out.println("正在生產數(shù)據(jù)...");
                Thread.sleep(random.nextInt(1000));
                data = "data" + count.incrementAndGet();
                System.out.println("將數(shù)據(jù)：" + data + "放入隊列...");
                if (!queue.offer(data,2, TimeUnit.SECONDS)){
                    System.out.println("數(shù)據(jù)放入失敗");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("生產者線程關閉");
            }
        }
    }

    public void stop(){
        isRunning = false;
    }
}

消費者線程

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: ruan
 * Date: 2021/9/25 17:48
 * @Description: 消費者線程
 */
public class Consumer implements Runnable {
    /**
     * 創(chuàng)建阻塞隊列
     */
    private BlockingQueue<String> queue;
    /**
     * 構造函數(shù)
     */
    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("消費者線程啟動...");
        Random random = new Random();
        boolean isRunning = true;
        while (isRunning){
            System.out.println("正在從隊列中獲取數(shù)據(jù)");
            try {
                String data = queue.poll(2, TimeUnit.SECONDS);
                if (null != data){
                    System.out.println("拿到數(shù)據(jù)：" + data);
                    System.out.println("正在消費數(shù)據(jù)：" + data);
                    Thread.sleep(random.nextInt(1000));
                }else {
                    //如果2s內無數(shù)據(jù)，認為生產者線程退出，消費者線程也退出
                    isRunning = false;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("消費者線程退出");
            }
        }
    }
}

主程序

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

/**
 * @Author: ruan
 * Date: 2021/9/25 17:56
 * @Description:
 */
public class Main {


    public static void main(String[] args) throws InterruptedException {
        /**
         * 創(chuàng)建阻塞隊列
         */
        BlockingQueue<String> queue = new LinkedBlockingDeque<>(10);

        /**
         * 創(chuàng)建生產者和消費者
         */
        Producer producer1 = new Producer(queue);
        Producer producer2 = new Producer(queue);
        Producer producer3 = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        /**
         * 創(chuàng)建線程
         */
        ExecutorService service = Executors.newCachedThreadPool();

        service.execute(producer1);
        service.execute(producer2);
        service.execute(producer3);
        service.execute(consumer);
        Thread.sleep(10000);
        //停止生產
        producer1.stop();
        producer2.stop();
        producer3.stop();
        Thread.sleep(2000);
        service.shutdown();
    }
}

DelayQueue

DelayQueue中的元素只有當其指定的延遲時間到了，才能夠從隊列中獲取到該元素。DelayQueue是一個沒有大小限制的隊列，因此往隊列中插入數(shù)據(jù)的操作（生產者）永遠不會被阻塞，而只有獲取數(shù)據(jù)的操作（消費者）才會被阻塞。

使用場景：

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

PriorityBlockingQueue

基于優(yōu)先級的阻塞隊列（優(yōu)先級的判斷通過構造函數(shù)傳入的Compator對象來決定），但需要注意的是PriorityBlockingQueue并不會阻塞數(shù)據(jù)生產者，而只會在沒有可消費的數(shù)據(jù)時，阻塞數(shù)據(jù)的消費者。因此使用的時候要特別注意，生產者生產數(shù)據(jù)的速度絕對不能快于消費者消費數(shù)據(jù)的速度，否則時間一長，會最終耗盡所有的可用堆內存空間。在實現(xiàn)PriorityBlockingQueue時，內部控制線程同步的鎖采用的是公平鎖。

SynchronousQueue

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

聲明一個SynchronousQueue有兩種不同的方式，它們之間有著不太一樣的行為。公平模式和非公平模式的區(qū)別:

如果采用公平模式：SynchronousQueue會采用公平鎖，并配合一個FIFO隊列來阻塞多余的生產者和消費者，從而體系整體的公平策略；

但如果是非公平模式（SynchronousQueue默認）：SynchronousQueue采用非公平鎖，同時配合一個LIFO隊列來管理多余的生產者和消費者，而后一種模式，如果生產者和消費者的處理速度有差距，則很容易出現(xiàn)饑渴的情況，即可能有某些生產者或者是消費者的數(shù)據(jù)永遠都得不到處理。

小結

BlockingQueue不光實現(xiàn)了一個完整隊列所具有的基本功能，同時在多線程環(huán)境下，他還自動管理了多線間的自動等待于喚醒功能，從而使得程序員可以忽略這些細節(jié)，關注更高級的功能。

到此這篇關于java阻塞隊列BlockingQueue詳細解讀的文章就介紹到這了,更多相關java阻塞隊列BlockingQueue內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論