《javaEE篇》--阻塞队列详解_业界新闻

发布时间:2024-08-03 04:33

阅读量:0

阻塞队列

阻塞队列概述

阻塞队列也是一种队列，和普通队列一样遵循先进先出的原则，但是阻塞队列相较于普通队列多了两项功能阻塞添加和阻塞移除，使得阻塞队列成为一种线程安全的数据结构

阻塞添加：当队列满的时候继续入队就会阻塞等待，直到有线程从队列中取走一个元素之后，队列才能重新添加新元素
阻塞移除：当队列为空时继续出队此时也会阻塞等待，直到有线程在队列中添加新的元素之后，队列才可以继续删除元素

生产者消费者模型

生产者消费者模型就是通过一个中间容器，来解决生产者和消费者之间的强耦合作用。

生产者和消费者不直接接触，而是由这个中间容器来完成二者之间的交流，生产者将生产的数据直接扔到中间容器里，消费者直接从中间容器里拿数据。

这里的中间容器就相当于是一个仓库，当生产者生产的数据太多时仓库满了，此时再向仓库内输入数据就会阻塞，相反当仓库为空时，消费者在向仓库索要数据，也会发生阻塞。这样就可以很好的平衡两者之间的处理能力。

生产者的生产速率和消费者的消费速率往往是不同的，难免会出现生产速率暴增或消费效率暴增的情况(比如各大电商平台的双11活动)，这时如果让生产者和消费者直接对接很有可能就会出现问题(比如服务器挂掉)，所以要使用这样一个中间容器来平衡，生产者可以按照自己的速率向容器中填充数据，消费者也可以按照自己的速率拿取数据。

阻塞队列的作用

解耦合

把阻塞队列封装成单独的服务器程序，将服务器a接收到的指令传给阻塞队列，再将指令从阻塞队列传到服务器b，此时，耦合就会被降低，如果b这边出现问题，就不会对a产生直接影响

削峰填谷

举个例子🌰：

假如在某电商平台的双11活动，此时用户发送的请求会突然暴增，如果服务器a收到的每一个请求都发给服务器b，那么，a和b所接触到的访问量就一样了，虽然ab的访问量一样，但是可能由于a和b上所跑的业务可能不一样，就会导致单个访问所消耗的硬件资源不一样，最终造成，A可以承受住这些访问量，但是b不可以。比如b要操作数据库，数据库本身就是分布式系统中相对脆弱的环节。

如果此时使用阻塞队列，A将收到的大量请求先写入到队列中，b就可以按照，之前的节奏来处理请求

削峰就相当于阻塞队列帮b承担了来自a的大量访问的压力，B可以按照之前的速度来进行处理，而且像上述峰值的情况一般，不会持续存在，只会短时间出现。
填谷就相当于，峰值之后a的请求量就恢复正常或者降低了，b就可以逐渐的把积压的数据都给处理掉

标准库中的阻塞队列

在 Java 标准库中内置了阻塞队列. 如果我们需要在一些程序中使用阻塞队列, 直接使用标准库中的即可.

LinkedBlockingQueue<> 链表实现的阻塞队列
LinkedTransferQueue<> 链表实现的无界阻塞队列
ArrayBlockingQueue<> 顺序表实现的有界阻塞队列
PriorityBlockingQueue<> 支持优先级排序的阻塞队列(由堆实现)
synchronousQueue<> 不存储元素的阻塞队列，每次入队操作必须伴随一个出队操作

//比较常用的就是 LinkedBlockingQueue<>

一般使用的BlockingQueue<>只是一个接口，真正实现的类是LinkedBlockingQueue<>

阻塞队列常用的阻塞入队操作是put，阻塞出队操作是take

实现阻塞队列

环形队列

博主这里使用数组环形队列来实现一个阻塞队列，环形队列是一种特殊的队列，在逻辑上是环形的数组首尾相连，但是实际上是一个定长的数组，环形队列依然保持队列先入先出的特点。

下面我通过画图来描绘一下

定义一个head表示队头，在定义一个tail表示队尾(实际上是指向队尾的下一个位置)，在初始状态下队列为空，head和tail重合

入队列操作就是，先在tail处插入新的元素，然后让tail++。出队列就是，把head位置的数据删除掉，直接让head++就行(逻辑删除),已经删除的数据会被之后插入的数据覆盖掉，当head和tail再一次重和就意味着队列为空了。数组内真正有效的数据就是区间[head,tail)内的数据。

但是这样做还有一个问题，当队列满的时候

此时head和tail就又重合了，队列空的时候他俩重合，队列满的时候他俩也是重合，这样似乎就不太好判断队列究竟是满还是空。这里有两种方法可以解决

浪费一个格子让tail指向head前一个时，队列就算满了
专门定义一个变量size来表示队列里的有效元素个数，当size=0时队列为空，size=”数组长度“时为满

两种方法都可以，这里我采用的是第二种方法

代码展示

public class AnnularQueue {     private String[] elems =new String[100];     //队首     private int head = 0;     //队尾     private int tail = 0;     //有效元素个数     private int size = 0;      //入队方法     public void put(String elem) throws InterruptedException {         //当队列已满时         if(size == elems.length) {             return;         }         //将元素插入队尾         elems[tail] = elem;         tail++;         //超过数组长度后回到数组首位         if(tail >= elems.length) {             tail = 0;         }         //有效元素总数加一         size++;     }      //出队方法     public String tack() throws InterruptedException {         String elem = null;         //当队列为空时         if(size == 0) {             return elem;         }         //出队(逻辑删除)         elem = elems[head];         head++;         if(head == elems.length) {             //回到数组首位             head = 0;         }         //存入后有效元素总数加一         size--;         //返回取出的元素         return elem;     }      }

实现阻塞队列

之前有提到阻塞队列的特点在于，当队列为满和为空时都会进行阻塞等待，所以我们只需要在刚刚实现的环形队列上加入这个功能就行。

代码展示

class MyBlockingQueue{     private String[] data = new String[1000];     private volatile int head = 0;     private volatile int tail = 0;     private volatile int size = 0;     public void put(String elem) throws InterruptedException {         synchronized (this){             //队列是否满了             //为了确认wait在唤醒之后，还可以再确认条件是否成立             while(size == data.length){                 //如果是普通队列就直接返回                 //return;                 //如果是阻塞队列就直接wait                 this.wait();             }             data[tail] = elem;             tail++;             size++;             //如果tail自增之后到达数组末尾，就让tail重新回到数组开头             if(tail == data.length){                 tail = 0;             }             //这个notify是用来唤醒因为队列为空而引起的阻塞             //take中的wait             this.notify();         }     }     public String take() throws InterruptedException {        synchronized (this){            //队列是否为空            //            while(size == 0){                //如果是普通队列就直接返回空                //return null;                //如果是阻塞队列就直接等待                this.wait();            }            //队列不为空，就去返回队首head处数据            String elem = data[head];            head++;            size--;            if(head == data.length){                head = 0;            }            //这个notify是用来唤醒，因队列满而阻塞的wait            //put中的wait            this.notify();            return  elem;        }     } }

既然我们要在队列满和空时让队列阻塞，那么就要使用wait来让调用该方法的线程等待阻塞

但是不能就这样一直等待下去，还要使用notify来解除等待，根据最开始的分析，当队列因为出队操作而阻塞时，就要等待其他线程进行入队操作，当队列因为入队操作阻塞时，就要等待其他线程进行出队操作。那么我们在tack和put操作最后各加上一个notify，这样就可以实现上面的功能了。因为如果队列处在阻塞状态下，必然是因为满了或者为空，而且必须是在其他线程完成tack/put操作之后，在可以使用notify解阻塞。
put的notify用来唤醒因为tack而引起的阻塞
tack的notify用来唤醒因为put而引起的阻塞

这里还有一个问题wait一定会被notify唤醒吗？或者说只有notify可以唤醒wait吗？

当然不止，还会因为interrupt方法直接中断当前的线程，不过我们刚刚使用了throws InterruptedException，会使线程直接报异常然后结束整个方法，这样是没问题的，但是如果使用try-catch就会处bug了。

此时方法并不会结束，而是会继续往下执行，如果队列已经满了的话，队列的最后一个元素就会被新插入的元素覆盖掉，但是被覆盖的元素并不是一个无效元素，而且此时size也比数组长度大，这样显然是不合理的。

所以在这种情况下我们要注意当前唤醒wait是notify还是interrupt，如果是notify证明此时队列已经不满可以继续插入元素，如果是interrupt唤醒，则此时队列还是瞒着的如果插入元素就会出现问题。

虽然我们刚刚使用的是throws直接抛出异常，这样就算是interrupt唤醒的也没什么大事，但是以防万一，我们还是要处理一下这个问题。

处理的方法也很简单，当wait被唤醒之后再判断一次队列是否为满，要是为满就继续wait，因为wait也可能被interrupt连续唤醒好几次，所以我们直接使用while作为判断语句，直到队列不满时才可以进行新的入队操作

到这里我们简单实现的一个阻塞队列就算完成了，接下来我们使用这个阻塞队列来实现一个简单的生产者消费者模型

实现生产者消费者模型

//实现生产者消费这模式         MyBlockingQueue blockingQueue = new MyBlockingQueue();         //生产者         Thread t1 = new Thread(() -> {             int num = 0;             while (true){                 try {                     //生产数据                     blockingQueue.put(num + "");//转换为字符                     System.out.println("生产：" + num);                     num++;                     Thread.sleep(1000);                 } catch (InterruptedException e) {                     throw new RuntimeException(e);                 }             }         });         //消费者         Thread t2 = new Thread(() -> {             while (true){                 try {                     //消费数据                     String elem = blockingQueue.take();                     System.out.println("消费：" + elem);                 } catch (InterruptedException e) {                     throw new RuntimeException(e);                 }             }         });         t1.start();         t2.start();