Java 阻塞隊(duì)列詳解及簡單使用
Java 阻塞隊(duì)列詳解
概要:
在新增的Concurrent包中,BlockingQueue很好的解決了多線程中,如何高效安全“傳輸”數(shù)據(jù)的問題。通過這些高效并且線程安全的隊(duì)列類,為我們快速搭建高質(zhì)量的多線程程序帶來極大的便利。本文詳細(xì)介紹了BlockingQueue家庭中的所有成員,包括他們各自的功能以及常見使用場景。
認(rèn)識BlockingQueue阻塞隊(duì)列,顧名思義,首先它是一個(gè)隊(duì)列,而一個(gè)隊(duì)列在數(shù)據(jù)結(jié)構(gòu)中所起的作用大致如下圖所示:
從上圖我們可以很清楚看到,通過一個(gè)共享的隊(duì)列,可以使得數(shù)據(jù)由隊(duì)列的一端輸入,從另外一端輸出;
常用的隊(duì)列主要有以下兩種:(當(dāng)然通過不同的實(shí)現(xiàn)方式,還可以延伸出很多不同類型的隊(duì)列,DelayQueue就是其中的一種)
1. 先進(jìn)先出(FIFO):先插入的隊(duì)列的元素也最先出隊(duì)列,類似于排隊(duì)的功能。從某種程度上來說這種隊(duì)列也體現(xiàn)了一種公平性。
2. 后進(jìn)先出(LIFO):后插入隊(duì)列的元素最先出隊(duì)列,這種隊(duì)列優(yōu)先處理最近發(fā)生的事件。
多線程環(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),萬一發(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é),尤其還要兼顧效率和線程安全,而這會給我們的程序帶來不小的復(fù)雜度。好在此時(shí),強(qiáng)大的concurrent包橫空出世了,而他也給我們帶來了強(qiáng)大的BlockingQueue。(在多線程領(lǐng)域:所謂阻塞,在某些情況下會掛起線程(即阻塞),一旦條件滿足,被掛起的線程又會自動被喚醒)
下面兩幅圖演示了BlockingQueue的兩個(gè)常見阻塞場景:
如上圖所示:當(dāng)隊(duì)列中沒有數(shù)據(jù)的情況下,消費(fèi)者端的所有線程都會被自動阻塞(掛起),直到有數(shù)據(jù)放入隊(duì)列。
如上圖所示:當(dāng)隊(duì)列中填滿數(shù)據(jù)的情況下,生產(chǎn)者端的所有線程都會被自動阻塞(掛起),直到隊(duì)列中有空的位置,線程被自動喚醒。
這也是我們在多線程環(huán)境下,為什么需要BlockingQueue的原因。作為BlockingQueue的使用者,我們再也不需要關(guān)心什么時(shí)候需要阻塞線程,什么時(shí)候需要喚醒線程,因?yàn)檫@一切BlockingQueue都給你一手包辦了。既然BlockingQueue如此神通廣大,讓我們一起來見識下它的常用方法:
BlockingQueue的核心方法:
放入數(shù)據(jù):
offer(anObject):表示如果可能的話,將anObject加到BlockingQueue里,即如果BlockingQueue可以容納, 則返回true,否則返回false
?。ū痉椒ú蛔枞?dāng)前執(zhí)行方法的線程)
offer(E o, long timeout, TimeUnit unit),可以設(shè)定等待的時(shí)間,如果在指定的時(shí)間內(nèi),還不能往隊(duì)列中加入BlockingQueue,則返回失敗。
put(anObject):把a(bǔ)nObject加到BlockingQueue里,如果BlockQueue沒有空間,則調(diào)用此方法的線程被阻斷直到BlockingQueue里面有空間再繼續(xù).
獲取數(shù)據(jù):
poll(time):取走BlockingQueue里排在首位的對象,若不能立即取出,則可以等time參數(shù)規(guī)定的時(shí)間,
取不到時(shí)返回null;
poll(long timeout, TimeUnit unit):從BlockingQueue取出一個(gè)隊(duì)首的對象,如果在指定時(shí)間內(nèi),
隊(duì)列一旦有數(shù)據(jù)可取,則立即返回隊(duì)列中的數(shù)據(jù)。否則知道時(shí)間超時(shí)還沒有數(shù)據(jù)可取,返回失敗。
take():取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻斷進(jìn)入等待狀態(tài)直到
BlockingQueue有新的數(shù)據(jù)被加入;
drainTo():一次性從BlockingQueue獲取所有可用的數(shù)據(jù)對象(還可以指定獲取數(shù)據(jù)的個(gè)數(shù)),
通過該方法,可以提升獲取數(shù)據(jù)效率;不需要多次分批加鎖或釋放鎖。
常見BlockingQueue
在了解了BlockingQueue的基本功能后,讓我們來看看BlockingQueue家庭大致有哪些成員?
BlockingQueue成員詳細(xì)介紹
1. ArrayBlockingQueue
基于數(shù)組的阻塞隊(duì)列實(shí)現(xiàn),在ArrayBlockingQueue內(nèi)部,維護(hù)了一個(gè)定長數(shù)組,以便緩存隊(duì)列中的數(shù)據(jù)對象,這是一個(gè)常用的阻塞隊(duì)列,除了一個(gè)定長數(shù)組外,ArrayBlockingQueue內(nèi)部還保存著兩個(gè)整形變量,分別標(biāo)識著隊(duì)列的頭部和尾部在數(shù)組中的位置。
ArrayBlockingQueue在生產(chǎn)者放入數(shù)據(jù)和消費(fèi)者獲取數(shù)據(jù),都是共用同一個(gè)鎖對象,由此也意味著兩者無法真正并行運(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í)不會產(chǎn)生或銷毀任何額外的對象實(shí)例,而后者則會生成一個(gè)額外的Node對象。這在長時(shí)間內(nèi)需要高效并發(fā)地處理大批量數(shù)據(jù)的系統(tǒng)中,其對于GC的影響還是存在一定的區(qū)別。而在創(chuàng)建ArrayBlockingQueue時(shí),我們還可以控制對象的內(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ì)列會從生產(chǎn)者手中獲取數(shù)據(jù),并緩存在隊(duì)列內(nèi)部,而生產(chǎn)者立即返回;只有當(dāng)隊(duì)列緩沖區(qū)達(dá)到最大值緩存容量時(shí)(LinkedBlockingQueue可以通過構(gòu)造函數(shù)指定該值),才會阻塞生產(chǎn)者隊(duì)列,直到消費(fèi)者從隊(duì)列中消費(fèi)掉一份數(shù)據(jù),生產(chǎn)者線程會被喚醒,反之對于消費(fèi)者這端的處理也基于同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理并發(fā)數(shù)據(jù),還因?yàn)槠鋵τ谏a(chǎn)者端和消費(fèi)者端分別采用了獨(dú)立的鎖來控制數(shù)據(jù)同步,這也意味著在高并發(fā)的情況下生產(chǎn)者和消費(fèi)者可以并行地操作隊(duì)列中的數(shù)據(jù),以此來提高整個(gè)隊(duì)列的并發(fā)性能。
作為開發(fā)者,我們需要注意的是,如果構(gòu)造一個(gè)LinkedBlockingQueue對象,而沒有指定其容量大小,LinkedBlockingQueue會默認(rèn)一個(gè)類似無限大小的容量(Integer.MAX_VALUE),這樣的話,如果生產(chǎn)者的速度一旦大于消費(fèi)者的速度,也許還沒有等到隊(duì)列滿阻塞產(chǎn)生,系統(tǒng)內(nèi)存就有可能已被消耗殆盡了。
ArrayBlockingQueue和LinkedBlockingQueue是兩個(gè)最普通也是最常用的阻塞隊(duì)列,一般情況下,在處理多線程間的生產(chǎn)者消費(fèi)者問題,使用這兩個(gè)類足以。
下面的代碼演示了如何使用BlockingQueue:
import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; /** * @author jackyuj */ 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); // 借助Executors ExecutorService service = Executors.newCachedThreadPool(); // 啟動線程 service.execute(producer1); service.execute(producer2); service.execute(producer3); service.execute(consumer); // 執(zhí)行10s Thread.sleep(10 * 1000); producer1.stop(); producer2.stop(); producer3.stop(); Thread.sleep(2000); // 退出Executor service.shutdown(); } } import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; /** * 消費(fèi)者線程 * * @author jackyuj */ public class Consumer implements Runnable { public Consumer(BlockingQueue<String> queue) { this.queue = queue; } public void run() { System.out.println("啟動消費(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)退出,自動退出消費(fèi)線程。 isRunning = false; } } } catch (InterruptedException e) { e.printStackTrace(); Thread.currentThread().interrupt(); } finally { System.out.println("退出消費(fèi)者線程!"); } } private BlockingQueue<String> queue; private static final int DEFAULT_RANGE_FOR_SLEEP = 1000; } import java.util.Random; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; /** * 生產(chǎn)者線程 * * @author jackyuj */ public class Producer implements Runnable { public Producer(BlockingQueue queue) { this.queue = queue; } public void run() { String data = null; Random r = new Random(); System.out.println("啟動生產(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; } private volatile boolean isRunning = true; private BlockingQueue queue; private static AtomicInteger count = new AtomicInteger(); private static final int DEFAULT_RANGE_FOR_SLEEP = 1000; }
1.DelayQueue
DelayQueue中的元素只有當(dāng)其指定的延遲時(shí)間到了,才能夠從隊(duì)列中獲取到該元素。DelayQueue是一個(gè)沒有大小限制的隊(duì)列,因此往隊(duì)列中插入數(shù)據(jù)的操作(生產(chǎn)者)永遠(yuǎn)不會被阻塞,而只有獲取數(shù)據(jù)的操作(消費(fèi)者)才會被阻塞。
使用場景:
DelayQueue使用場景較少,但都相當(dāng)巧妙,常見的例子比如使用一個(gè)DelayQueue來管理一個(gè)超時(shí)未響應(yīng)的連接隊(duì)列。
2.PriorityBlockingQueue
基于優(yōu)先級的阻塞隊(duì)列(優(yōu)先級的判斷通過構(gòu)造函數(shù)傳入的Compator對象來決定),但需要注意的是PriorityBlockingQueue并不會阻塞數(shù)據(jù)生產(chǎn)者,而只會在沒有可消費(fèi)的數(shù)據(jù)時(shí),阻塞數(shù)據(jù)的消費(fèi)者。因此使用的時(shí)候要特別注意,生產(chǎn)者生產(chǎn)數(shù)據(jù)的速度絕對不能快于消費(fèi)者消費(fèi)數(shù)據(jù)的速度,否則時(shí)間一長,會最終耗盡所有的可用堆內(nèi)存空間。在實(shí)現(xiàn)PriorityBlockingQueue時(shí),內(nèi)部控制線程同步的鎖采用的是公平鎖。
3.SynchronousQueue
一種無緩沖的等待隊(duì)列,類似于無中介的直接交易,有點(diǎn)像原始社會中的生產(chǎn)者和消費(fèi)者,生產(chǎn)者拿著產(chǎn)品去集市銷售給產(chǎn)品的最終消費(fèi)者,而消費(fèi)者必須親自去集市找到所要商品的直接生產(chǎn)者,如果一方?jīng)]有找到合適的目標(biāo),那么對不起,大家都在集市等待。相對于有緩沖的BlockingQueue來說,少了一個(gè)中間經(jīng)銷商的環(huán)節(jié)(緩沖區(qū)),如果有經(jīng)銷商,生產(chǎn)者直接把產(chǎn)品批發(fā)給經(jīng)銷商,而無需在意經(jīng)銷商最終會將這些產(chǎn)品賣給那些消費(fèi)者,由于經(jīng)銷商可以庫存一部分商品,因此相對于直接交易模式,總體來說采用中間經(jīng)銷商的模式會吞吐量高一些(可以批量買賣);但另一方面,又因?yàn)榻?jīng)銷商的引入,使得產(chǎn)品從生產(chǎn)者到消費(fèi)者中間增加了額外的交易環(huán)節(jié),單個(gè)產(chǎn)品的及時(shí)響應(yīng)性能可能會降低。
聲明一個(gè)SynchronousQueue有兩種不同的方式,它們之間有著不太一樣的行為。公平模式和非公平模式的區(qū)別:
如果采用公平模式:SynchronousQueue會采用公平鎖,并配合一個(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)都得不到處理。
小結(jié)
BlockingQueue不光實(shí)現(xiàn)了一個(gè)完整隊(duì)列所具有的基本功能,同時(shí)在多線程環(huán)境下,他還自動管理了多線間的自動等待于喚醒功能,從而使得程序員可以忽略這些細(xì)節(jié),關(guān)注更高級的功能。
感謝閱讀,希望能幫助到大家,謝謝大家對本站的支持!
相關(guān)文章
Java中語音url轉(zhuǎn)換成InputStream的示例代碼
在Java中,可以使用java.net.URL和java.net.URLConnection類來將語音URL轉(zhuǎn)換為InputStream,本文通過示例代碼介紹Java中語音url轉(zhuǎn)換成InputStream的相關(guān)知識,感興趣的朋友一起看看吧2024-01-01Java讀取Oracle大字段數(shù)據(jù)(CLOB)的2種方法
這篇文章主要介紹了Java讀取Oracle大字段數(shù)據(jù)(CLOB)的2種方法,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-04-04Java中遍歷集合的并發(fā)修改異常解決方案實(shí)例代碼
當(dāng)你遍歷集合的同時(shí),又往集合中添加或者刪除元素,就可能報(bào)并發(fā)修改異常,下面這篇文章主要給大家介紹了關(guān)于Java中遍歷集合的并發(fā)修改異常解決方案的相關(guān)資料,需要的朋友可以參考下2022-12-12Java日期操作方法工具類實(shí)例【包含日期比較大小,相加減,判斷,驗(yàn)證,獲取年份等】
這篇文章主要介紹了Java日期操作方法工具類,結(jié)合完整實(shí)例形式分析了java針對日期的各種常見操作,包括日期比較大小,相加減,判斷,驗(yàn)證,獲取年份、天數(shù)、星期等,需要的朋友可以參考下2017-11-11如何集成swagger2構(gòu)建Restful API
這篇文章主要介紹了如何集成swagger2構(gòu)建Restful API,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11Springboot整合nacos報(bào)錯(cuò)無法連接nacos的解決
這篇文章主要介紹了Springboot整合nacos報(bào)錯(cuò)無法連接nacos的解決方案,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-06-06