java大厂面试题整理（四）阻塞队列

队列Queue本身其实就是一种数据结构。不管是学JVM还是学数据结构队列都是一个必不可少的概念。
当然了一般多和栈Stack一起讲。其区别就是队列先进先出。栈先进后出。
这里如果不是很明确的可以如我一般记：队列的重点是队，也就是排队的队。一般我们买票排队，是不是排在前面的先买票然后走啊。
而栈，用我们电脑的操作系统很容易想象到：我们打开某磁盘，一层一层往下走，当我们回退上一步的时候是退回当前的上一步。这也就是先点的在最外层。也可以理解为后来居上。

栈的理解

因为我们这篇笔记主要是说阻塞队列，所以这篇文章单纯说下队列。

队列

从宏观上说队列。队列在java中是Queue（顶层）
而其有两大分支：
一类是阻塞队列。一类是非阻塞队列。
其他的比如双端队列，优先队列什么的也都可以分为以上2类。
而我们要说的阻塞队列的核心就是下面框起来的这个：

阻塞队列核心接口

阻塞队列

阻塞队列本身其实也是队列。这个毫无疑问。只不过它的特点是阻塞。什么叫阻塞呢？
其实意思我们可以根据名字猜测：就是不是立即可取，而是排队候着。
其中有两点会涉及到阻塞：

1. 当队列中是空的，从队列获取元素会被阻塞。
2. 当队列中满了，往队列中添加元素会被阻塞。

比如说我们去银行办业务。一个人需要办理半小时。一共就俩窗口办理。第一个人来了，去1号窗口办理了。第二个人来了，去2号窗口办理了。这个时候第三个人来了。重点来了！肯定不会让这第三个人滚。正常情况下第三个人应该是取个号，然后在大厅等着。这个时候第四个人又来了，继续取号等着。
这个3，4又不给办理又不给撵走。就给搁大厅等着。这个大厅其实可以看做阻塞队列。排队的人就是阻塞队列中的元素。
当然了，因为时间有限。来三五个人都可以大厅等着。假如来三五十个人眼看今天不可能排队到，正常的大堂经理肯定不会都让等着了，到了某个限制节点，就应该说今天办不上了明天来吧合理劝退。这就是阻塞队列中队列满了阻塞。
当然了大厅一个人都没有，也没人办理业务。这两个人工窗口里的人也得在那坐着等人，这种情况也算是阻塞。

按照我们上面看Queue层级的时候说了阻塞队列的核心接口是：BlockingQueue
我们简单分析一下类的层级结构：

Queue和List和Set是兄弟
阻塞队列简单类结构
其中这7个阻塞队列的实现类简单介绍下：

1. ArrayBlockingQueue:由数组结构组成的有界阻塞队列（不指定默认是10）
2. LinkedBlockingQueue:由链表结构组成的有界（但大小默认为Integer最大值）阻塞队列（！！！！千万注意不能用默认的）
3. PriorityBlockingQueue：支持优先级排序的无界阻塞队列
4. SynchronousQueue:不存储元素的阻塞队列。也就是单个元素的队列
5. DelayQueue：使用优先级队列实现的延迟无界阻塞队列
6. LinkedTransferQueue：由链表结构组成的无界阻塞队列
7. LinkedBlockingDeque：由链表结构组成的双端阻塞队列

注意上面加重的三个是比较常用的，也是线程池使用的。剩下的其实大概就能明白了。这里只简单说下这三种。
我们通过对list，set的使用应该大概有个了解：数据结构的本身我们并不是很关心，主要在工作中还是增删查（如果是list可能还存在改的现象，但是set本身就是无序的，所以不存在改。我们逻辑上想修改也只能删除旧的，添加新的。）那么说到队列的基本操作也就是这三个：增删查。

阻塞队列的增删查
不同于list 的单纯的add/remove。阻塞队列的插入和移除有多组操作：
注意我这里说的是组。也就是插入和删除（还有检查）是一组。
首先说下第一组：add/remove。这个是我们比较常见的。但是当无可操作以后（满了的add和空值的remove）就会直接抛异常。
满了以后的add抛异常
空元素后的remove抛异常
这两段代码直接告诉我们add/remove的特性：无可操作抛异常
这组操作还有一个element（）方法，是检查当前队首元素
第二组：offer/poll。队列满了再存返回false。队列空了再移除返回null。
下面是代码操作：
满了再offer返回false，空着poll返回null
第三组：put/take。这一组操作是阻塞的。也就是满了再put就卡在那等着。空了再take就一直等有元素了再移除。
下面是代码示例：
put是阻塞的
注意看执行到第四个put的时候就卡在那了，线程一直在等待。其实take也是同理。
无可take一直等着
第四组：offer(E,时间，单位)/poll(时间,单位)定时等待。到了时间还没等到就过了
这个其实方法名类似第2组，功能类似第三组。也算是二，三的结合了。既不直接返回，也不死死等待。如下代码示例：
注意看时间差，正好5s
同理移除也是这样：
移除也是阻塞了5s
因为没有添加元素，所以第一个移除就阻塞了。
然后这四组api的使用就都敲了一遍，其实都比较简单。多用用也就能记住了。而且为什么有四组呢？其实这个就是一个比较有意思的事了。可以从类结构说起。比如put/remove就是继承自Collection（算是集合的顶级接口类了吧）。剩下的三个则继承自BlockingQueue。阻塞队列的核心接口类。然后我们可以根据不同的情况来使用。
接下来我们把上面七个阻塞队列的实现类中重点的三个实际应用一下：
SynchronousQueue：
这个东西咋说呢，谁用谁知道。上文说了，不存储元素，也就是当你第一个put/add/poll的时候，就已经是阻塞住了（可以想象成容积是0.但凡往里放都是满了。会阻塞）。直到有人来取就直接带走阻塞这元素了。就是把这个队列绕过去了。重点是不存储元素。而是产生一个消费一个。
继续做比喻:如果说银行工作人员是取。去银行办理业务的是存。银行大厅可以容纳10人，这个10人其实就是可以预存的，也就是队列的长度。
然后工作人员按顺序办理其实就是按顺序取。取没工作人员就干等着。超过10个人要进来就劝退。
但是这个SynchronousQueue完全改套路了。银行关门了。你要是想办理啥银行门口蹲着吧。反正这么大地方只能蹲一个人。然后如果有工作人员看到你就偷偷帮你把业务办了。我们其实去银行也是为了找工作人员。其实如果工作人员在大街上，我们去大街也行。工作人员在铁轨上我们去铁轨也行。
重点是办理业务的人员和工作人员对接上就行。银行不是必须的。
所以这个SynchronousQueue队列存储元素不是必须的。只要存取都阻塞，能凑成一对一对的就行了被！

阻塞队列用在哪里？

其实这个比上文中的介绍还要重要。毕竟只是主要还是为了使用的。阻塞队列常用的几个地方:

• 生产者消费者模式（这个是一种设计模式。下面两个才算是实际应用。）
• 线程池
• 消息中间件

生产者消费者模式

其实这个是很常用的一个模式。生产出一个东西才可以用一个。生活中生产消费的例子:早餐店->做出一屉包子->可以卖一屉包子。
做出是生产。卖是消费。如果有人买没做好可以等一会儿做好了再买。如果做好了没人买可以先放着等客人来再卖。
这个是一个很好理解的逻辑。而我们的代码中生产者消费者模式就是类似的逻辑。
但是他的实现不是像包子这样简单。而是有一个进化的过程。我们从最原始的实现开始说。

---

这里要说一个口诀（据说阳哥的精辟总结）：
线程 操作 资源类
判断 干活 通知
防止虚假唤醒

---

简单解释下：java中万物皆对象。所以其实并发中本质就是不同的线程想对同一个对象去做一些事。
比如说一个空调温度25°。你觉得冷想打高1°。别人就想25°。那么你和别人就是两个线程。资源类就是空调。操作就是+1/-1.
而判断干活通知也是一个流程：首先你要看当前温度是不是25°。是的话你要打高一度。但是如果当前已经26了，你就不用了。 判断是25则打高一度是干活。 当你打高一度以后，你就告诉别人，我已经把空调打高一度了（意思是别人可以再低一度了。）下面是一个代码demo。
题目：一个初始值为0的遍历。两个线程交替操作。一个+1.一个-1.来五轮!
第一版最原始的代码:

/**
 * 
 * @author lisijia
 *
 */
public class Test9 {
    volatile int i = 0;
    public synchronized void add() {
        try {
            //判断（这里用while而不是while是为了防止虚假唤醒）
            while(i != 0) {
                //条件不符合，不操作
                wait();
            }
            //注意：判断过了才能走到这   干活
            i++;
            System.out.println(Thread.currentThread().getName()+"执行了加法操作！");
            //通知
            notifyAll();
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
    public synchronized void sub(){
        try {
            //判断  
            while(i != 1) {
                //条件不符合，不操作
                wait();
            }
            //注意：判断过了才能走到这   干活
            i--;
            System.out.println(Thread.currentThread().getName()+"执行了减法操作！");
            //通知
            notifyAll();
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
    public static void main(String[] args) throws Exception {
        Test9 test9 = new Test9();
        for(int i = 0;i<5;i++) {
            new Thread(()->{
                test9.add();
            }).start();
            new Thread(()->{
                test9.sub();
            }).start();
        }
        
    }

}

第二版进化版代码:(主要是lock替代synchronized)

/**
 * 
 * @author lisijia
 *
 */
public class Test9 {
    volatile int i = 0;
    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();
    public void add() {
        lock.lock();
        try {
            //判断（这里用while而不是while是为了防止虚假唤醒）
            while(i != 0) {
                //条件不符合，不操作
                condition.await();
            }
            //注意：判断过了才能走到这   干活
            i++;
            System.out.println(Thread.currentThread().getName()+"执行了加法操作！");
            //通知
            condition.signalAll();
        } catch (Exception e) {
            // TODO: handle exception
        }finally {
            lock.unlock();
        }
    }
    public void sub(){
        lock.lock();
        try {
            //判断（这里用while而不是while是为了防止虚假唤醒）
            while(i != 1) {
                //条件不符合，不操作
                condition.await();
            }
            //注意：判断过了才能走到这   干活
            i--;
            System.out.println(Thread.currentThread().getName()+"执行了减法操作！");
            //通知
            condition.signalAll();
        } catch (Exception e) {
            // TODO: handle exception
        }finally {
            lock.unlock();
        }
    }
    public static void main(String[] args) throws Exception {
        Test9 test9 = new Test9();
        for(int i = 0;i<5;i++) {
            new Thread(()->{
                test9.add();
            }).start();
            new Thread(()->{
                test9.sub();
            }).start();
        }   
    }
}

所谓的虚假唤醒，就是是指在多线程的情况下同时唤醒多个线程，但是其实只有一个线程是有用的唤醒。打个比方：进货是生产者，卖货是消费者。当我们只进了一件货的时候会唤醒所有的消费者，比如唤醒了100个消费者。但是本质山只能有一个人买到这件货。所以其它的都算是虚假唤醒。会不会有人疑问明明都加锁了为什么还会出现不同的结果呢？本质就因为一句话：！！！不管是wait还是await。唤醒之后都是直接从wait/await后面继续往下执行！
因为if只会执行一次，执行完会接着向下执行if（）外边的（所以多线程可能0的是时候-1操作停止了。但是别的-1以后变成0唤醒了这个线程。然后继续往下执行，0-1出现了-1这种奇怪的现象。。。）
而while不会，直到条件满足才会向下执行while（）外边的
并且这里重点说下第一版和第二版的进阶区别:
其实上面模板中主要三个功能:

1. 加锁
2. 判断条件不满足释放当前锁
3. 唤醒其余线程

而第一版本上面三个功能实现分别是：synchronized，wait，notify
而第二个版本的实现分别是：lock，await，singal
而synchronized和Lock到底有什么区别？为什么说是进化了呢？

1. synchronized是关键字，属于JVM的。用的monitor。 同时只有在被synchronized修饰的同步块或者方法中才能调用wait/notify。
   Lock是一个具体的类。是api层面的锁。
2. synchronized不需要用户去手动释放锁。当synchronized代码执行完毕后会自动让线程释放锁。
   ReentrantLock需要手动去加锁释放锁。若没有主动释放锁则可能产生死锁现象。
3. 等待是否可种断？
   synchronized不可中断，除非抛异常或者正常运行完毕。
   ReentrantLock可中断。可设置获取锁的超时时间（等待xx时间，到了时间还没等到就不等了）。
4. 是否是公平锁。synchronized一定是非公平锁。而Lock虽然默认不设置是非公平锁，但是是可以设置为公平锁的。
5. 锁要绑定多个条件的Condition
   Synchronized没有
   ReentrantLock用来实现分组唤醒的线程们，也都可以精确唤醒。而synchronized要么随机唤醒一个要么唤醒全部。
   比如如下一个题目：
   一个线程输出A。一个线程输出B。一个线程输出C。问如何实现输出AABBCCC如此循环下去。
   其实这个题除了多条件，和之前那道题差不多。而且昨天那个模板也可以实现这个功能。只不过这个针对唤醒的实现本质上会省去好多无用的唤醒。尤其是当线程比较多的时候，下面是一个简单的实现代码：

/**
 * 
 * @author lisijia
 *
 */
public class Test10 {
    
    Lock lock = new ReentrantLock();
    Condition condition1 = lock.newCondition();
    Condition condition2 = lock.newCondition();
    Condition condition3 = lock.newCondition();
    private int i = 1;
    public void A() {
        lock.lock();
        try {
            while (i > 5) {
                condition1.await();             
            }
            System.out.println("A");
            i++;
            if(i == 5)condition2.signal();
        }catch (Exception e) {
            // TODO: handle exception
        } finally {
            lock.unlock();
        }
    }
    public void B() {
        lock.lock();
        try {
            while(i>15 || i<6) {
                condition2.await();             
            }
            System.out.println("B");
            i++;
            if(i == 15) condition3.signal();
        }catch (Exception e) {
            // TODO: handle exception
        } finally {
            lock.unlock();
        }
    }
    public void C() {
        lock.lock();
        try {
            while(i<15) {
                condition3.await();             
            }
            System.out.println("C");
            i++;
            if(i == 31) {
                i = 1;
                condition1.signal();
            }
        }catch (Exception e) {
            // TODO: handle exception
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        Test10 t = new Test10();
        new Thread(()->{
            for(int i = 0;i<10;i++) {
                t.A();
            }
        }).start();
        new Thread(()->{
            for(int i = 0;i<100;i++) {
                t.B();
            }
        }).start();
        new Thread(()->{
            for(int i = 0;i<100;i++) {
                t.C();
            }
        }).start();

    }

}

依旧是按照口诀来的：线程操作资源类。然后判断干活通知。只不过通知的时候是针对通知的了。
第三版代码也就是现在主要讲的：阻塞队列实现：

public class Test11 {
    
    volatile boolean flag = true;
    AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<String> blockingQueue = null;
    public Test11(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }
    public void myProd() throws Exception{
        String data = null;
        boolean retValue;
        while(flag) {
            TimeUnit.SECONDS.sleep(1l);
            data = atomicInteger.incrementAndGet()+"";
            retValue = blockingQueue.offer(data,2l,TimeUnit.SECONDS);
            if(retValue) {
                System.out.println("插入"+data+"成功！");
            }
        }
        System.out.println("已经叫停，不再插入！");
    }
    
    public void myConsumer() throws Exception{
        String result = null;
        while(flag) {
            result = blockingQueue.poll(2,TimeUnit.SECONDS);
            if(result == null) {
                flag = false;//到这说明队列里没内容了。所以直接叫停
                System.out.println("等待时间过长，强制停止！");             
            }else {
                System.out.println("消费"+result+"成功！");
            }
        }
        System.out.println("停止消费！");
    }
    public void stop() {
        this.flag = false;
    }
    public static void main(String[] args) throws Exception {
        Test11 test11 = new Test11(new ArrayBlockingQueue<String>(10));
        new Thread(()->{
            try {
                test11.myProd();
            } catch (Exception e) {
                // TODO: handle exception
            }
        }).start();
        new Thread(()->{
            try {
                test11.myConsumer();
            } catch (Exception e) {
                // TODO: handle exception
            }
        }).start();
        TimeUnit.SECONDS.sleep(5l);
        test11.stop();
    }   
}

生产者消费者模式虽然没有加锁，但是也实现了让线程按顺序执行，如上代码运行结果：

运行结果

一定是生产一个消费一个的。而且我们可以用atomicInteger 计数来判断停止条件是否满足（上面一二版代码直接用i++是因为代码加锁了。但是这个版本因为没有锁，一定要用原子类计数 ！！！）
其实我觉得这个不能完全说生产者消费者模式就代替了锁，毕竟都有其独特的作用和功能。但是生产者消费者模式相对于锁而言性能提高，代码可以更灵活，当然了某些场景还要要用锁的。存在即合理。只不过我们要发散思路，多知道一些总能让我们遇到问题多一些解决办法。

这篇笔记就记到这里，如果稍微帮到你了记得点个喜欢点个关注，也祝大家工作顺顺利利！