面試官：你手寫過(guò)堵塞隊(duì)列嗎？我懵了。。

閱讀本文大概需要 5 分鐘。

來(lái)自：blog.csdn.net/qq_38306425/article/details/109332045

面試官：你好，你先做個(gè)自我介紹吧

某人：面試官你好，我叫開(kāi)局一張嘴面試全靠吹，某某年畢業(yè)，畢業(yè)自家里蹲大學(xué)，做過(guò)某某項(xiàng)目。。。。。。

面試官微微一笑,捋了捋稀疏的頭發(fā):看你簡(jiǎn)歷，你精通多線程？那你手寫過(guò)堵塞隊(duì)列嗎？

某人心里出現(xiàn)一萬(wàn)個(gè)問(wèn)號(hào)，堵塞隊(duì)列是啥玩意？平時(shí)基本都是crud，頂多用多線程跑數(shù)據(jù)

某人：沒(méi)有手寫過(guò)。

面試官：哦，那你說(shuō)下堵塞隊(duì)列吧

某人支支吾吾：這個(gè)有點(diǎn)忘了

面試官：沒(méi)事，那我們下一個(gè)。

此處省略一萬(wàn)字。

面試官扭了扭嚴(yán)重負(fù)荷的頸椎：先到這里吧，你先回去等通知。

某人：好的。

不出意外，某人等了一個(gè)月，等的望眼欲穿，也沒(méi)等到那個(gè)期待的電話。

1.什么是隊(duì)列

隊(duì)列是一種特殊的線性表，特殊之處在于它只允許在表的前端（front）進(jìn)行刪除操作，而在表的后端（rear）進(jìn)行插入操作，和棧一樣，隊(duì)列是一種操作受限制的線性表。進(jìn)行插入操作的端稱為隊(duì)尾，進(jìn)行刪除操作的端稱為隊(duì)頭。

隊(duì)列其實(shí)就是跟平時(shí)排隊(duì)一樣，按照順序來(lái)，先排隊(duì)的先買到東西，后排隊(duì)的后買到東西，排隊(duì)的第一個(gè)叫隊(duì)頭，最后一個(gè)叫隊(duì)尾，這就是隊(duì)列的先進(jìn)先出，這是和棧最大的區(qū)別。

推薦下自己做的 Spring Boot 的實(shí)戰(zhàn)項(xiàng)目：

https://github.com/YunaiV/ruoyi-vue-pro

2.什么是堵塞隊(duì)列?

當(dāng)隊(duì)列為空時(shí)，消費(fèi)者掛起，隊(duì)列已滿時(shí)，生產(chǎn)者掛起，這就是生產(chǎn)-消費(fèi)者模型，堵塞其實(shí)就是將線程掛起。因?yàn)樯a(chǎn)者的生產(chǎn)速度和消費(fèi)者的消費(fèi)速度之間的不匹配，就可以通過(guò)堵塞隊(duì)列讓速度快的暫時(shí)堵塞, 如生產(chǎn)者每秒生產(chǎn)兩個(gè)數(shù)據(jù)，而消費(fèi)者每秒消費(fèi)一個(gè)數(shù)據(jù)，當(dāng)隊(duì)列已滿時(shí)，生產(chǎn)者就會(huì)堵塞（掛起），等待消費(fèi)者消費(fèi)后，再進(jìn)行喚醒。

堵塞隊(duì)列會(huì)通過(guò)掛起的方式來(lái)實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者之間的平衡，這是和普通隊(duì)列最大的區(qū)別。

推薦下自己做的 Spring Cloud 的實(shí)戰(zhàn)項(xiàng)目：

https://github.com/YunaiV/onemall

3.如何實(shí)現(xiàn)堵塞隊(duì)列？

jdk其實(shí)已經(jīng)幫我們提供了實(shí)現(xiàn)方案，java5增加了concurrent包，concurrent包中的BlockingQueue就是堵塞隊(duì)列，我們不需要關(guān)心BlockingQueue如何實(shí)現(xiàn)堵塞，一切都幫我們封裝好了，只需要做一個(gè)沒(méi)有感情的API調(diào)用者就行。

4.BlockingQueue如何使用？

BlockingQueue本身只是一個(gè)接口，規(guī)定了堵塞隊(duì)列的方法，主要依靠幾個(gè)實(shí)現(xiàn)類實(shí)現(xiàn)。

4.1 BlockingQueue主要方法

1.插入數(shù)據(jù) （1）offer(E e)：如果隊(duì)列沒(méi)滿，返回true，如果隊(duì)列已滿，返回false（不堵塞） （2）offer(E e, long timeout, TimeUnit unit)：可以設(shè)置等待時(shí)間，如果隊(duì)列已滿，則進(jìn)行等待。超過(guò)等待時(shí)間，則返回false （3）put(E e)：無(wú)返回值，一直等待，直至隊(duì)列空出位置

2.獲取數(shù)據(jù) （1）poll()：如果有數(shù)據(jù)，出隊(duì)，如果沒(méi)有數(shù)據(jù)，返回null （2）poll(long timeout, TimeUnit unit)：可以設(shè)置等待時(shí)間，如果沒(méi)有數(shù)據(jù)，則等待，超過(guò)等待時(shí)間，則返回null （3）take()：如果有數(shù)據(jù)，出隊(duì)。如果沒(méi)有數(shù)據(jù)，一直等待（堵塞）

4.2 BlockingQueue主要實(shí)現(xiàn)類

1.ArrayBlockingQueue：ArrayBlockingQueue是基于數(shù)組實(shí)現(xiàn)的，通過(guò)初始化時(shí)設(shè)置數(shù)組長(zhǎng)度，是一個(gè)有界隊(duì)列，而且ArrayBlockingQueue和LinkedBlockingQueue不同的是，ArrayBlockingQueue只有一個(gè)鎖對(duì)象，而LinkedBlockingQueue是兩個(gè)鎖對(duì)象，一個(gè)鎖對(duì)象會(huì)造成要么是生產(chǎn)者獲得鎖，要么是消費(fèi)者獲得鎖，兩者競(jìng)爭(zhēng)鎖，無(wú)法并行。

2.LinkedBlockingQueue：LinkedBlockingQueue是基于鏈表實(shí)現(xiàn)的，和ArrayBlockingQueue不同的是，大小可以初始化設(shè)置，如果不設(shè)置，默認(rèn)設(shè)置大小為Integer.MAX_VALUE，LinkedBlockingQueue有兩個(gè)鎖對(duì)象，可以并行處理。

3.DelayQueue：DelayQueue是基于優(yōu)先級(jí)的一個(gè)無(wú)界隊(duì)列，隊(duì)列元素必須實(shí)現(xiàn)Delayed接口，支持延遲獲取，元素按照時(shí)間排序，只有元素到期后，消費(fèi)者才能從隊(duì)列中取出。

4.PriorityBlockingQueue：PriorityBlockingQueue是基于優(yōu)先級(jí)的一個(gè)無(wú)界隊(duì)列，底層是基于數(shù)組存儲(chǔ)元素的，元素按照優(yōu)選級(jí)順序存儲(chǔ)，優(yōu)先級(jí)是通過(guò)Comparable的compareTo方法來(lái)實(shí)現(xiàn)的（自然排序），和其他堵塞隊(duì)列不同的是，其只會(huì)堵塞消費(fèi)者，不會(huì)堵塞生產(chǎn)者，數(shù)組會(huì)不斷擴(kuò)容，這就是一個(gè)彩蛋，使用時(shí)要謹(jǐn)慎。

5.SynchronousQueue：SynchronousQueue是一個(gè)特殊的隊(duì)列，其內(nèi)部是沒(méi)有容器的，所以生產(chǎn)者生產(chǎn)一個(gè)數(shù)據(jù)，就堵塞了，必須等消費(fèi)者消費(fèi)后，生產(chǎn)者才能再次生產(chǎn)，稱其為隊(duì)列有點(diǎn)不合適，現(xiàn)實(shí)生活中，多個(gè)人才能稱為隊(duì)，一個(gè)人稱為隊(duì)有些說(shuō)不過(guò)去。

5.手寫堵塞隊(duì)列

我是參照了ArrayBlockingQueue的源碼寫的，歡迎大家斧正。

/**
?*?@author?yz
?*?@version?1.0
?*?@date?2020/10/31?11:24
?*/
????public?class?YzBlockingQuery?{

????private?Object[]?tab;?//隊(duì)列容器

????private?int?takeIndex;?//出隊(duì)下標(biāo)

????private?int?putIndex;?//入隊(duì)下標(biāo)

????private?int?size;//元素?cái)?shù)量

????private?ReentrantLock?reentrantLock?=?new?ReentrantLock();

????private?Condition?notEmpty;//讀條件

????private?Condition?notFull;//寫條件

????public?YzBlockingQuery(int?tabCount)?{
????????if?(tabCount?<=?0)?{
????????????new?NullPointerException();
????????}

????????tab?=?new?Object[tabCount];
????????notEmpty?=?reentrantLock.newCondition();
????????notFull?=?reentrantLock.newCondition();
????}

????public?boolean?offer(Object?obj)?{
????????if?(obj?==?null)?{?throw?new?NullPointerException();?}
????????try?{
????????????//獲取鎖
????????????reentrantLock.lock();
????????????//隊(duì)列已滿
????????????while?(size==tab.length){
????????????????System.out.println("隊(duì)列已滿");
????????????????//堵塞
????????????????notFull.await();
????????????}
????????????tab[putIndex]=obj;
????????????if(++putIndex==tab.length){
????????????????putIndex=0;
????????????}
????????????size++;
????????????//喚醒讀線程
????????????notEmpty.signal();
????????????return?true;
????????}?catch?(Exception?e)?{
????????????//喚醒讀線程
????????????notEmpty.signal();
????????}?finally?{
????????????reentrantLock.unlock();
????????}
????????return?false;
????}


????public?Object?take(){
????????try?{
????????????reentrantLock.lock();
????????????while?(size==0){
????????????????System.out.println("隊(duì)列空了");
????????????????//堵塞
????????????????notEmpty.await();
????????????}
????????????Object?obj=?tab[takeIndex];
????????????//如果到了最后一個(gè)，則從頭開(kāi)始
????????????if(++takeIndex==tab.length){
????????????????takeIndex=0;
????????????}
????????????size--;
????????????//喚醒寫線程
????????????notFull.signal();
????????????return?obj;
????????}catch?(Exception?e){
????????????//喚醒寫線程
????????????notFull.signal();
????????}finally?{
????????????reentrantLock.unlock();
????????}
????????return?null;
????}


????public?static?void?main(String[]?args)?{
????????Random?random?=?new?Random(100);
????????YzBlockingQuery?yzBlockingQuery=new?YzBlockingQuery(5);
????????Thread?thread1?=?new?Thread(()?->?{
????????????for?(int?i=0;i<100;i++)?{
????????????????try?{
????????????????????Thread.sleep(300);
????????????????}?catch?(InterruptedException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????????yzBlockingQuery.offer(i);
????????????????System.out.println("生產(chǎn)者生產(chǎn)了："+i);
????????????}
????????});
????
????????Thread?thread2?=?new?Thread(()?->?{
????????????for?(int?i=0;i<100;i++)?{
????????????????try?{
????????????????????Thread.sleep(1000);
????????????????}?catch?(InterruptedException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????????Object?take?=?yzBlockingQuery.take();
????????????????System.out.println("消費(fèi)者消費(fèi)了："+take);
????????????}
????????});
????
????????thread1.start();
????????thread2.start();
????}
}