什么是阻塞隊(duì)列？

阻塞隊(duì)列（BlockingQueue）是一個(gè)支持兩個(gè)附加操作的隊(duì)列。這兩個(gè)附加的操作是：在隊(duì)列為空時(shí)，獲取元素的線程會(huì)等待隊(duì)列變?yōu)榉强铡．?dāng)隊(duì)列滿時(shí)，存儲(chǔ)元素的線程會(huì)等待隊(duì)列可用。阻塞隊(duì)列常用于生產(chǎn)者和消費(fèi)者的場(chǎng)景，生產(chǎn)者是往隊(duì)列里添加元素的線程，消費(fèi)者是從隊(duì)列里拿元素的線程。阻塞隊(duì)列就是生產(chǎn)者存放元素的容器，而消費(fèi)者也只從容器里拿元素。

阻塞隊(duì)列提供了四種處理方法:

方法\處理方式	拋出異常	返回特殊值	一直阻塞	超時(shí)退出
插入	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除	remove()	poll()	take()	poll(time,unit)
檢查	element()	peek()	不可用	不可用

?

?

?

?

?

• 拋出異常：是指當(dāng)阻塞隊(duì)列滿時(shí)候，再往隊(duì)列里插入元素，會(huì)拋出IllegalStateException("Queue full")異常。當(dāng)隊(duì)列為空時(shí)，從隊(duì)列里獲取元素時(shí)會(huì)拋出NoSuchElementException異常 。
• 返回特殊值：插入方法會(huì)返回是否成功，成功則返回true。移除方法，則是從隊(duì)列里拿出一個(gè)元素，如果沒有則返回null
• 一直阻塞：當(dāng)阻塞隊(duì)列滿時(shí)，如果生產(chǎn)者線程往隊(duì)列里put元素，隊(duì)列會(huì)一直阻塞生產(chǎn)者線程，直到拿到數(shù)據(jù)，或者響應(yīng)中斷退出。當(dāng)隊(duì)列空時(shí)，消費(fèi)者線程試圖從隊(duì)列里take元素，隊(duì)列也會(huì)阻塞消費(fèi)者線程，直到隊(duì)列可用。
• 超時(shí)退出：當(dāng)阻塞隊(duì)列滿時(shí)，隊(duì)列會(huì)阻塞生產(chǎn)者線程一段時(shí)間，如果超過一定的時(shí)間，生產(chǎn)者線程就會(huì)退出

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

BlockingQueue成員詳細(xì)介紹

1. ArrayBlockingQueue

?

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

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

2. LinkedBlockingQueue

基于鏈表的阻塞隊(duì)列，同ArrayListBlockingQueue類似，其內(nèi)部也維持著一個(gè)數(shù)據(jù)緩沖隊(duì)列（該隊(duì)列由一個(gè)鏈表構(gòu)成），當(dāng)生產(chǎn)者往隊(duì)列中放入一個(gè)數(shù)據(jù)時(shí)，隊(duì)列會(huì)從生產(chǎn)者手中獲取數(shù)據(jù)，并緩存在隊(duì)列內(nèi)部，而生產(chǎn)者立即返回；只有當(dāng)隊(duì)列緩沖區(qū)達(dá)到最大值緩存容量時(shí)（LinkedBlockingQueue可以通過構(gòu)造函數(shù)指定該值），才會(huì)阻塞生產(chǎn)者隊(duì)列，直到消費(fèi)者從隊(duì)列中消費(fèi)掉一份數(shù)據(jù)，生產(chǎn)者線程會(huì)被喚醒，反之對(duì)于消費(fèi)者這端的處理也基于同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理并發(fā)數(shù)據(jù)，還因?yàn)槠鋵?duì)于生產(chǎn)者端和消費(fèi)者端分別采用了獨(dú)立的鎖來控制數(shù)據(jù)同步，這也意味著在高并發(fā)的情況下生產(chǎn)者和消費(fèi)者可以并行地操作隊(duì)列中的數(shù)據(jù)，以此來提高整個(gè)隊(duì)列的并發(fā)性能。
作為開發(fā)者，我們需要注意的是，如果構(gòu)造一個(gè)LinkedBlockingQueue對(duì)象，而沒有指定其容量大小，LinkedBlockingQueue會(huì)默認(rèn)一個(gè)類似無(wú)限大小的容量（Integer.MAX_VALUE），這樣的話，如果生產(chǎn)者的速度一旦大于消費(fèi)者的速度，也許還沒有等到隊(duì)列滿阻塞產(chǎn)生，系統(tǒng)內(nèi)存就有可能已被消耗殆盡了。

ArrayBlockingQueue和LinkedBlockingQueue是兩個(gè)最普通也是最常用的阻塞隊(duì)列，一般情況下，在處理多線程間的生產(chǎn)者消費(fèi)者問題，使用這兩個(gè)類足以。

示例：

生產(chǎn)者

?

package cn.com.example.concurrent.blockingQueue;import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger;/*** Created by Jack on 2017/1/24.*/ 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("啟動(dòng)生產(chǎn)者線程！");try {while (isRunning) {System.out.println("正在生產(chǎn)數(shù)據(jù)...");Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));data = "data:" + count.incrementAndGet();System.out.println("將數(shù)據(jù)：" + data + "放入隊(duì)列...");if (!queue.offer(data, 2, TimeUnit.SECONDS)) {System.out.println("放入數(shù)據(jù)失敗：" + data);}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出生產(chǎn)者線程！");}}public void stop() {isRunning = false;} }

?

消費(fèi)者

package cn.com.example.concurrent.blockingQueue;import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit;/*** Created by Jack on 2017/1/24.*/ 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("啟動(dòng)消費(fèi)者線程！");Random r = new Random();boolean isRunning = true;try {while (isRunning) {System.out.println("正從隊(duì)列獲取數(shù)據(jù)...");String data = queue.poll(2, TimeUnit.SECONDS);if (null != data) {System.out.println("拿到數(shù)據(jù)：" + data);System.out.println("正在消費(fèi)數(shù)據(jù)：" + data);Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));} else {// 超過2s還沒數(shù)據(jù)，認(rèn)為所有生產(chǎn)線程都已經(jīng)退出，自動(dòng)退出消費(fèi)線程。isRunning = false;}}} catch (InterruptedException e) {e.printStackTrace();Thread.currentThread().interrupt();} finally {System.out.println("退出消費(fèi)者線程！");}} }

測(cè)試

package cn.com.example.concurrent.blockingQueue;import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue;/*** Created by Jack on 2017/1/24.*/ public class BlockingQueueTest {public static void main(String[] args) throws InterruptedException {// 聲明一個(gè)容量為10的緩存隊(duì)列BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);Producer producer1 = new Producer(queue);Producer producer2 = new Producer(queue);Producer producer3 = new Producer(queue);Consumer consumer = new Consumer(queue);// 借助ExecutorsExecutorService service = Executors.newCachedThreadPool();// 啟動(dòng)線程service.execute(producer1);service.execute(producer2);service.execute(producer3);service.execute(consumer);// 執(zhí)行10sThread.sleep(10 * 1000);producer1.stop();producer2.stop();producer3.stop();Thread.sleep(2000);// 退出Executorservice.shutdown();}}

結(jié)果

啟動(dòng)消費(fèi)者線程！ 正從隊(duì)列獲取數(shù)據(jù)... 啟動(dòng)生產(chǎn)者線程！ 啟動(dòng)生產(chǎn)者線程！ 正在生產(chǎn)數(shù)據(jù)... 啟動(dòng)生產(chǎn)者線程！ 正在生產(chǎn)數(shù)據(jù)... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:1放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 拿到數(shù)據(jù)：data:1 正在消費(fèi)數(shù)據(jù)：data:1 將數(shù)據(jù)：data:2放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 正從隊(duì)列獲取數(shù)據(jù)... 拿到數(shù)據(jù)：data:2 正在消費(fèi)數(shù)據(jù)：data:2 將數(shù)據(jù)：data:3放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:4放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:5放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:6放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:7放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 正從隊(duì)列獲取數(shù)據(jù)... 拿到數(shù)據(jù)：data:3 正在消費(fèi)數(shù)據(jù)：data:3 將數(shù)據(jù)：data:8放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:9放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:10放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 正從隊(duì)列獲取數(shù)據(jù)... 拿到數(shù)據(jù)：data:4 正在消費(fèi)數(shù)據(jù)：data:4 將數(shù)據(jù)：data:11放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:12放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:13放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:14放入隊(duì)列... 正在生產(chǎn)數(shù)據(jù)... 將數(shù)據(jù)：data:15放入隊(duì)列...

3. DelayQueue

DelayQueue中的元素只有當(dāng)其指定的延遲時(shí)間到了，才能夠從隊(duì)列中獲取到該元素。DelayQueue是一個(gè)沒有大小限制的隊(duì)列，因此往隊(duì)列中插入數(shù)據(jù)的操作（生產(chǎn)者）永遠(yuǎn)不會(huì)被阻塞，而只有獲取數(shù)據(jù)的操作（消費(fèi)者）才會(huì)被阻塞。
使用場(chǎng)景：
　　DelayQueue使用場(chǎng)景較少，但都相當(dāng)巧妙，常見的例子比如使用一個(gè)DelayQueue來管理一個(gè)超時(shí)未響應(yīng)的連接隊(duì)列。

4. PriorityBlockingQueue

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

5. SynchronousQueue

一種無(wú)緩沖的等待隊(duì)列，類似于無(wú)中介的直接交易，有點(diǎn)像原始社會(huì)中的生產(chǎn)者和消費(fèi)者，生產(chǎn)者拿著產(chǎn)品去集市銷售給產(chǎn)品的最終消費(fèi)者，而消費(fèi)者必須親自去集市找到所要商品的直接生產(chǎn)者，如果一方?jīng)]有找到合適的目標(biāo)，那么對(duì)不起，大家都在集市等待。相對(duì)于有緩沖的BlockingQueue來說，少了一個(gè)中間經(jīng)銷商的環(huán)節(jié)（緩沖區(qū)），如果有經(jīng)銷商，生產(chǎn)者直接把產(chǎn)品批發(fā)給經(jīng)銷商，而無(wú)需在意經(jīng)銷商最終會(huì)將這些產(chǎn)品賣給那些消費(fèi)者，由于經(jīng)銷商可以庫(kù)存一部分商品，因此相對(duì)于直接交易模式，總體來說采用中間經(jīng)銷商的模式會(huì)吞吐量高一些（可以批量買賣）；但另一方面，又因?yàn)榻?jīng)銷商的引入，使得產(chǎn)品從生產(chǎn)者到消費(fèi)者中間增加了額外的交易環(huán)節(jié)，單個(gè)產(chǎn)品的及時(shí)響應(yīng)性能可能會(huì)降低。
　　聲明一個(gè)SynchronousQueue有兩種不同的方式，它們之間有著不太一樣的行為。公平模式和非公平模式的區(qū)別:
　　如果采用公平模式：SynchronousQueue會(huì)采用公平鎖，并配合一個(gè)FIFO隊(duì)列來阻塞多余的生產(chǎn)者和消費(fèi)者，從而體系整體的公平策略；
　　但如果是非公平模式（SynchronousQueue默認(rèn)）：SynchronousQueue采用非公平鎖，同時(shí)配合一個(gè)LIFO隊(duì)列來管理多余的生產(chǎn)者和消費(fèi)者，而后一種模式，如果生產(chǎn)者和消費(fèi)者的處理速度有差距，則很容易出現(xiàn)饑渴的情況，即可能有某些生產(chǎn)者或者是消費(fèi)者的數(shù)據(jù)永遠(yuǎn)都得不到處理。

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

?

轉(zhuǎn)載于:https://www.cnblogs.com/Zombie-Xian/p/6347386.html

總結(jié)

以上是生活随笔為你收集整理的java.util.concurrent BlockingQueue详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。