当前位置：首页 > news >正文

JavaEE多线程案例之阻塞队列

news 2025/7/4 4:59:10

上文我们了解了多线程案例中的单例模式，此文我们来探讨多线程案例之阻塞队列吧

1. 阻塞队列是什么？

阻塞队列是⼀种特殊的队列.也遵守"先进先出"的原则.
阻塞队列是⼀种线程安全的数据结构,并且具有以下特性:

当队列满的时候,继续⼊队列就会阻塞,直到有其他线程从队列中取⾛元素.
当队列空的时候,继续出队列也会阻塞,直到有其他线程往队列中插⼊元素.

阻塞队列的⼀个典型应用场景就是"生产者消费者模型".这是⼀种非常典型的开发模型

那么什么是生产者消费者模型呢？？

1.1 生产者消费者模型

1. 生产者消费者模式就是通过⼀个容器来解决生产者和消费者的强耦合问题

生产者和消费者彼此之间不直接通讯，而是通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找⽣产者要数据，而是直接从阻塞队列⾥取.

应用场景：
一对夫妻开包子店，早上夫妻二人包包子，女的负责擀包子皮（擀好的包子皮放在桌板上（类似于阻塞队列）），男的负责包包子，其中擀包子皮的就是生产者，包包子的就是消费者。擀包子皮的人不在意谁消耗他生产出来的包子皮，包包子的人也不在意是谁生产的包子皮，能用就可以

2. 阻塞队列就相当于⼀个缓冲区，平衡了⽣产者和消费者的处理能⼒. (削峰填⾕）

应用场景：
在每年的双十一、双十二购物节，服务器同⼀时刻可能会收到大量的支付请求，如果同时处理这些支付请求服务器可能扛不住(每个支付请求的处理都需要比较复杂的流程).
这个时候就可以把这些请求都放到⼀个阻塞队列中(类似于上面提到的放包子皮的桌板，消费者不需要按照生产者的请求速度来完成，可以按照自己的速度来完成，就保证了不会服务器崩掉),
然后再由消费者线程慢慢的来处理每个⽀付请求.这样做可以有效进⾏ “削峰”, 防⽌服务器被突然到来的⼀波请求直接冲垮.

3. 阻塞队列可以实现异步操作

1.2 标准库中的阻塞队列

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

BlockingQueue 是⼀个接口. 真正实现的类是 LinkedBlockingQueue.
put 方法用于阻塞式的入队列, take 用于阻塞式的出队列.
BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性.

在这里插入图片描述

1.3 消息队列的自我实现

首先我们采用的是循环队列的方式实现消息队列

定义一个数组用于存放消息
定义一个头尾指针标记消息队列的队头和队尾
定义一个size用于标记有效数据个数

//定义一个消息队列数组
private Integer [] information=null;
//定义一个头指针
private int head=0;
//尾指针
private int tail=0;
//定义一个size用于标记有效数据个数
private int size=0;

此时我们数组不给定大小，我们采用构造方法在初始化时，给定数组大小，此时和JDK实现的雷同

//构造方法用于初始化消息队列的容量大小
public MyBlockQueue(Integer capacity) {if(capacity<=0){throw new RuntimeException("队列容量必须大于0.");}information=new Integer[capacity];
}

在这里插入图片描述

在队尾插入元素

为了实现消息队列的效果，就要与普通队列不一样在插入元素时，
若消息队列满了，就等待，等到消息队列有空位时，就会被唤醒
当插入元素时，就唤醒出元素的的操作
其中，wait()方法、notifyAll()需要搭配synchronize关键字使用

public void put(int value) throws InterruptedException {//没有位置就要等待if(size>=information.length){synchronized (this){this.wait();}}//有位置直接插入元素，无需等待if(size<information.length){information[tail]=value;tail++;size++;if(tail>=information.length){tail=0;}synchronized (this){this.notifyAll();}}}

在队头出元素

为了实现消息队列的效果，当消息队列中没有元素时，就需要等待，直到有元素进来，再被唤醒
当出元素时，有位置空余出来，就可以唤醒要插入的操作

//取出元素public int take() throws InterruptedException {//队列为空，需要等待if(size==0){synchronized (this){this.wait();}}//有元素直接取出即可Integer value=information[head];head++;size--;if(head==information.length){head=0;}synchronized (this){this.notifyAll();}return value;}

synchronize关键字实现了原子性、在效果上实现了内存可见性，但是没有保证有序性，所以我们为涉及修改的变量加上volatile关键字

    //定义一个消息队列数组private volatile Integer [] information=null;//定义一个头指针private volatile int head=0;//尾指针private volatile int tail=0;//定义一个size用于标记有效数据个数private volatile int  size=0;

由于在插入和删除的操作中涉及多个变量的修改，我们可以扩大synchronize的范围

此时，我们完整的代码如下：

public class MyBlockQueue {//定义一个消息队列数组private volatile Integer [] information=null;//定义一个头指针private volatile int head=0;//尾指针private volatile int tail=0;//定义一个size用于标记有效数据个数private volatile int  size=0;//构造方法用于初始化消息队列的容量大小public MyBlockQueue(Integer capacity) {if(capacity<=0){throw new RuntimeException("队列容量必须大于0.");}information=new Integer[capacity];}//插入元素public void put(int value) throws InterruptedException {synchronized (this){//没有位置就要等待if(size>=information.length){this.wait();}//有位置直接插入元素，无需等待if(size<information.length){information[tail]=value;tail++;size++;if(tail>=information.length){tail=0;}synchronized (this){this.notifyAll();}}}}//取出元素public int take() throws InterruptedException {synchronized (this){//队列为空，需要等待if(size==0){synchronized (this){this.wait();}}//有元素直接取出即可Integer value=information[head];head++;size--;if(head==information.length){head=0;}this.notifyAll();return value;}}
}

此时引发一个新问题，假设此时有多个线程要进行put操作，但只有一个空余位置，会有什么问题吗

在这里插入图片描述
所以我们将if判断改成while判断，并且在JDK提供的put方法中也是用while

1.4 消息队列代码

public class MyBlockQueue {//定义一个消息队列数组private volatile Integer [] information=null;//定义一个头指针private volatile int head=0;//尾指针private volatile int tail=0;//定义一个size用于标记有效数据个数private volatile int  size=0;//构造方法用于初始化消息队列的容量大小public MyBlockQueue(Integer capacity) {if(capacity<=0){throw new RuntimeException("队列容量必须大于0.");}information=new Integer[capacity];}//插入元素public void put(int value) throws InterruptedException {synchronized (this){//此处最好使⽤ while.//否则notifyAll 的时候, 该线程从wait 中被唤醒,  但是紧接着并未抢占到锁.//当锁被抢占的时候, 可能⼜已经队列满了while (size>=information.length){this.wait();}information[tail]=value;tail++;size++;if (tail>information.length){tail=0;}this.notifyAll();}}//取出元素public int take() throws InterruptedException {synchronized (this){//队列为空，需要等待while (size==0){this.wait();}//有元素直接取出即可Integer value=information[head];head++;size--;if(head==information.length){head=0;}this.notifyAll();return value;}}
}