手把手教你寫堵塞隊列！

面試官：你好，你先做個自我介紹吧

某人：面試官你好，我叫開局一張嘴面試全靠吹，某某年畢業(yè)，畢業(yè)自家里蹲大學，做過某某項目。。。。。。

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

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

某人：沒有手寫過。

面試官：哦，那你說下堵塞隊列吧

某人支支吾吾：這個有點忘了

面試官：沒事，那我們下一個。

此處省略一萬字。

面試官扭了扭嚴重負荷的頸椎：先到這里吧，你先回去等通知。

某人：好的。

不出意外，某人等了一個月，等的望眼欲穿，也沒等到那個期待的電話。

1.什么是隊列

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

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

2.什么是堵塞隊列?

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

堵塞隊列會通過掛起的方式來實現(xiàn)生產(chǎn)者和消費者之間的平衡，這是和普通隊列最大的區(qū)別。

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

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

4.BlockingQueue如何使用？

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

4.1 BlockingQueue主要方法

1.插入數(shù)據(jù)

（1）offer(E e)：如果隊列沒滿，返回true，如果隊列已滿，返回false（不堵塞）

（2）offer(E e, long timeout, TimeUnit unit)：可以設置等待時間，如果隊列已滿，則進行等待。超過等待時間，則返回false

（3）put(E e)：無返回值，一直等待，直至隊列空出位置

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

（1）poll()：如果有數(shù)據(jù)，出隊，如果沒有數(shù)據(jù)，返回null

（2）poll(long timeout, TimeUnit unit)：可以設置等待時間，如果沒有數(shù)據(jù)，則等待，超過等待時間，則返回null

（3）take()：如果有數(shù)據(jù)，出隊。如果沒有數(shù)據(jù)，一直等待（堵塞）

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

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

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

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

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

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

5.手寫堵塞隊列

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

/**  
 * @author yz  
 * @version 1.0   
 */  
    public class YzBlockingQuery {  
  
    private Object[] tab; //隊列容器  
  
    private int takeIndex; //出隊下標  
  
    private int putIndex; //入隊下標  
  
    private int size;//元素數(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();  
            //隊列已滿  
            while (size==tab.length){  
                System.out.println("隊列已滿");  
                //堵塞  
                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("隊列空了");  
                //堵塞  
                notEmpty.await();  
            }  
            Object obj= tab[takeIndex];  
            //如果到了最后一個，則從頭開始  
            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("消費者消費了："+take);  
            }  
        });  
      
        thread1.start();  
        thread2.start();  
    }  
}