最近在研究一些并發(fā)方面的技術(shù)其中研究到阻塞隊列（BlockingQueue）的時候做了寫筆記文檔 大家可以一起探討一下 ：

BlockingQueue，是  java.util.concurrent  包提供的用于解決并發(fā)生產(chǎn)者 - 消費者問題的最有用的類。

它的特性是在任意時刻只有一個線程可以進(jìn)行take或者put操作，并且BlockingQueue提供了超時return null的機制，在許多生產(chǎn)場景里都可以看到這個工具的身影。

一、隊列類型

1.無限隊列 （unbounded queue ） - 幾乎可以無限增長

2.有限隊列 （ bounded queue ） - 定義了最大容量

二、隊列數(shù)據(jù)結(jié)構(gòu)

隊列實質(zhì)就是一種存儲數(shù)據(jù)的結(jié)構(gòu)

通常用鏈表或者數(shù)組實現(xiàn)

一般而言隊列具備FIFO先進(jìn)先出的特性，當(dāng)然也有雙端隊列（Deque）優(yōu)先級隊列

主要操作：入隊（EnQueue）與出隊（Dequeue）

三、常見的4中阻塞隊列

• ArrayBlockingQueue 由數(shù)組支持的有界隊列
• LinkedBlockingQueue 由鏈接節(jié)點支持的可選有界隊列
• PriorityBlockingQueue 由優(yōu)先級堆支持的無界優(yōu)先級隊列
• DelayQueue 由優(yōu)先級堆支持的、基于時間的調(diào)度隊列

1.ArrayBlockingQueue

隊列基于數(shù)組實現(xiàn),容量大小在創(chuàng)建ArrayBlockingQueue對象時已定義好數(shù)據(jù)結(jié)構(gòu)如下圖：

①隊列創(chuàng)建：

 BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>();

②應(yīng)用場景：

在線程池中有比較多的應(yīng)用，生產(chǎn)者消費者場景

③工作原理：

基于ReentrantLock保證線程安全，根據(jù)Condition實現(xiàn)隊列滿時的阻塞

2.LinkedBlockingQueue

是一個基于鏈表的無界隊列(理論上有界)

BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>();

上面這段代碼中，blockingQueue 的容量將設(shè)置為 Integer.MAX_VALUE 。

向無限隊列添加元素的所有操作都將永遠(yuǎn)不會阻塞，[注意這里不是說不會加鎖保證線程安全]，因此它可以增長到非常大的容量。

使用無限 BlockingQueue 設(shè)計生產(chǎn)者 - 消費者模型時最重要的是 消費者應(yīng)該能夠像生產(chǎn)者向隊列添加消息一樣快地消費消息 。

否則，內(nèi)存可能會填滿，然后就會得到一個 OutOfMemory 異常。

3.DelayQueue 

由優(yōu)先級堆支持的、基于時間的調(diào)度隊列，內(nèi)部基于無界隊列PriorityQueue實現(xiàn)，而無界隊列基于數(shù)組的擴容實現(xiàn)。

①隊列創(chuàng)建：

 BlockingQueue<String> blockingQueue = new DelayQueue();      

②要求：

入隊的對象必須要實現(xiàn)Delayed接口,而Delayed集成自Comparable接口

③應(yīng)用場景：

電影票

④工作原理：

隊列內(nèi)部會根據(jù)時間優(yōu)先級進(jìn)行排序。延遲類線程池周期執(zhí)行。

4.BlockingQueue API

BlockingQueue 接口的所有方法可以分為兩大類：負(fù)責(zé)向隊列添加元素的方法和檢索這些元素的方法。在隊列滿/空的情況下，來自這兩個組的每個方法的行為都不同。

①添加元素

②檢索元素

在構(gòu)建生產(chǎn)者 - 消費者程序時，這些方法是 BlockingQueue 接口中最重要的構(gòu)建塊。

四、多線程生產(chǎn)者-消費者示例

接下來我們創(chuàng)建一個由兩部分組成的程序 - 生產(chǎn)者 ( Producer ) 和消費者 ( Consumer ) 。

生產(chǎn)者將生成一個 0 到 100 的隨機數(shù)(十全大補丸的編號)，并將該數(shù)字放在 BlockingQueue 中。

我們將創(chuàng)建 16 個線程（潘金蓮）用于生成隨機數(shù)并使用 put() 方法阻塞，直到隊列中有可用空間。

需要記住的重要一點是，我們需要阻止我們的消費者線程無限期地等待元素出現(xiàn)在隊列中。

從生產(chǎn)者(潘金蓮)向消費者(武大郎)發(fā)出信號的好方法是，不需要處理消息，而是發(fā)送稱為毒 （ poison ） 丸 （ pill ） 的特殊消息。

我們需要發(fā)送盡可能多的毒 （ poison ） 丸 （ pill ） ，因為我們有消費者(武大郎)。

然后當(dāng)消費者從隊列中獲取特殊的毒 （ poison ） 丸 （ pill ）消息時，它將優(yōu)雅地完成執(zhí)行。

以下生產(chǎn)者的代碼：

@Slf4j
public class NumbersProducer implements Runnable {
private final int poisonPill;
private final int poisonPillPerProducer;
private BlockingQueue<Integer> numbersQueue;
public NumbersProducer(BlockingQueue<Integer> numbersQueue,
int poisonPill,
int poisonPillPerProducer) {
this.numbersQueue = numbersQueue;
this.poisonPill = poisonPill;
this.poisonPillPerProducer = poisonPillPerProducer;
  }
@Override
  public void run() {
try {
      generateNumbers ();
    } catch (InterruptedException e) {
Thread.currentThread ().interrupt ();
    }
  }
private void generateNumbers() throws InterruptedException {
for (int i = 0; i < 100; i++) {
numbersQueue.put (ThreadLocalRandom.current ().nextInt (100));
      log.info ("潘金蓮-{}號,給武大郎的泡藥！", Thread.currentThread ().getId ());
    }
for (int j = 0; j < poisonPillPerProducer; j++) {
numbersQueue.put (poisonPill);
      log.info ("潘金蓮-{}號,往武大郎的藥里放入第{}顆毒丸！", Thread.currentThread ().getId (), j + 1);
    }
  }
}

我們的生成器構(gòu)造函數(shù)將 BlockingQueue 作為參數(shù)，用于協(xié)調(diào)生產(chǎn)者和使用者之間的處理。

我們看到方法 generateNumbers() 將 100 個元素（生產(chǎn)100副藥給武大郎吃）放入隊列中。

它還需要有毒 （ poison ） 丸 （ pill ） （潘金蓮給武大郎下毒）消息，以便知道在執(zhí)行完成時放入隊列的消息類型。

該消息需要將 poisonPillPerProducer 次放入隊列中。

每個消費者將使用 take() 方法從 BlockingQueue 獲取一個元素，因此它將阻塞，直到隊列中有一個元素。

從隊列中取出一個 Integer 后，它會檢查該消息是否是毒 （ poison ） 丸 （ pill ）（武大郎看潘金蓮有沒有下毒） ，如果是，則完成一個線程的執(zhí)行。

否則，它將在標(biāo)準(zhǔn)輸出上打印出結(jié)果以及當(dāng)前線程的名稱。

@Slf4j
public class NumbersConsumer implements Runnable {
private final int poisonPill;
private BlockingQueue<Integer> queue;
public NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
this.queue = queue;
this.poisonPill = poisonPill;
  }
@Override
  public void run() {
try {
while (true) {
Integer number = queue.take ();
if (number.equals (poisonPill)) {
return;
        }
        log.info ("武大郎-{}號,喝藥-編號:{}", Thread.currentThread ().getId (), number);
      }
    } catch (InterruptedException e) {
Thread.currentThread ().interrupt ();
    }
  }
}

需要注意的重要事項是隊列的使用。

與生成器構(gòu)造函數(shù)中的相同，隊列作為參數(shù)傳遞。

我們可以這樣做，是因為 BlockingQueue 可以在線程之間共享而無需任何顯式同步。

既然我們有生產(chǎn)者和消費者，我們就可以開始我們的計劃。我們需要定義隊列的容量，并將其設(shè)置為 10個元素。

我們創(chuàng)建4 個生產(chǎn)者線程，并且創(chuàng)建等于可用處理器數(shù)量的消費者線程：

public class Main {
public static void main(String[] args) {
int BOUND = 10;
int N_PRODUCERS = 16;
int N_CONSUMERS = Runtime.getRuntime ().availableProcessors ();
int poisonPill = Integer.MAX_VALUE;
int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
int mod = N_CONSUMERS % N_PRODUCERS;
    BlockingQueue<Integer> queue = new LinkedBlockingQueue<> (BOUND);
//潘金蓮給武大郎熬藥
    for (int i = 1; i < N_PRODUCERS; i++) {
new Thread (new NumbersProducer (queue, poisonPill, poisonPillPerProducer)).start ();
    }
//武大郎開始喝藥
    for (int j = 0; j < N_CONSUMERS; j++) {
new Thread (new NumbersConsumer (queue, poisonPill)).start ();
    }
//潘金蓮開始投毒，武大郎喝完毒藥GG
    new Thread (new NumbersProducer (queue, poisonPill, poisonPillPerProducer + mod)).start ();
  }
}

BlockingQueue 是使用具有容量的構(gòu)造創(chuàng)建的。我們正在創(chuàng)造 4 個生產(chǎn)者和 N 個消費者（武大郎）。

我們將我們的毒 （ poison ） 丸 （ pill ）消息指定為 Integer.MAX_VALUE，因為我們的生產(chǎn)者在正常工作條件下永遠(yuǎn)不會發(fā)送這樣的值。

這里要注意的最重要的事情是 BlockingQueue 用于協(xié)調(diào)它們之間的工作。

到此這篇關(guān)于一文簡介Java中BlockingQueue阻塞隊列的文章就介紹到這了,更多相關(guān)Java BlockingQueue阻塞隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論