阻塞队列
阻塞队列概述
阻塞队列也是一种队列,和普通队列一样遵循先进先出的原则,但是阻塞队列相较于普通队列多了两项功能阻塞添加和阻塞移除,使得阻塞队列成为一种线程安全的数据结构
- 阻塞添加:当队列满的时候继续入队就会阻塞等待,直到有线程从队列中取走一个元素之后,队列才能重新添加新元素
- 阻塞移除:当队列为空时继续出队此时也会阻塞等待,直到有线程在队列中添加新的元素之后,队列才可以继续删除元素
生产者消费者模型
生产者消费者模型就是通过一个中间容器,来解决生产者和消费者之间的强耦合作用。
生产者和消费者不直接接触,而是由这个中间容器来完成二者之间的交流,生产者将生产的数据直接扔到中间容器里,消费者直接从中间容器里拿数据。
这里的中间容器就相当于是一个仓库,当生产者生产的数据太多时仓库满了,此时再向仓库内输入数据就会阻塞,相反当仓库为空时,消费者在向仓库索要数据,也会发生阻塞。这样就可以很好的平衡两者之间的处理能力。
生产者的生产速率和消费者的消费速率往往是不同的,难免会出现生产速率暴增或消费效率暴增的情况(比如各大电商平台的双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;//如果是阻塞队列就直接waitthis.wait();}data[tail] = elem;tail++;size++;//如果tail自增之后到达数组末尾,就让tail重新回到数组开头if(tail == data.length){tail = 0;}//这个notify是用来唤醒因为队列为空而引起的阻塞//take中的waitthis.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中的waitthis.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();
//可以通过sleep来调整生产速率和消费速率
运行结果
可以看到我刚刚的代码是消费速率远大于生产速率(生产速率是每秒生产1),所以结果会按照生产速率来执行.
接着我们调整速率,这次是生产速率是按照编译器执行的速度,消费速率是每0.5秒消费1
可以看到生产者瞬间生产了1000个数据,但是消费者任然按照自己的速率去执行,这里阻塞队列就起到了一个削峰填谷的作用
以上就是博主对阻塞队列知识的分享,如果有不懂的或者有其他见解的欢迎在下方评论或者私信博主,也希望多多支持博主之后和博客!!🥰🥰