【JavaEE初阶】多线程 _ 基础篇 _ 阻塞队列（案例二）写在前面一、阻塞队列概论1.1 阻塞队列的概念与作用1.2 阻塞队列的应用场景 —— 生产者消费者模型1.3 生产者消费者模型的好处二、阻塞队列的实现2.1 Java标准库里面的实现2.2 自己动手去模拟实现一个阻塞队列2.2.1 首先实现一个普通队列2.2.2 接着需要解决线程安全问题2.2.3 最后来实现阻塞效果

46 阅读 0 评论 31 点赞

我是靠谱客的博主无私鞋垫，最近开发中收集的这篇文章主要介绍【JavaEE初阶】多线程 _ 基础篇 _ 阻塞队列（案例二）写在前面一、阻塞队列概论1.1 阻塞队列的概念与作用1.2 阻塞队列的应用场景 —— 生产者消费者模型1.3 生产者消费者模型的好处二、阻塞队列的实现2.1 Java标准库里面的实现2.2 自己动手去模拟实现一个阻塞队列2.2.1 首先实现一个普通队列2.2.2 接着需要解决线程安全问题2.2.3 最后来实现阻塞效果，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

☕导航小助手☕

????写在前面

????一、阻塞队列概论

????????1.1 阻塞队列的概念与作用

????????1.2 阻塞队列的应用场景 —— 生产者消费者模型

????????1.3 生产者消费者模型的好处

????二、阻塞队列的实现

????????2.1 Java标准库里面的实现

????????2.2 自己动手去模拟实现一个阻塞队列

????????????2.2.1 首先实现一个普通队列

????????????2.2.2 接着需要解决线程安全问题

????????????2.2.3 最后来实现阻塞效果

写在前面

在单例模式之后，我们就来学习一下第二个多线程案例 —— 阻塞队列~

如果学过数据结构的话，就会知道在数据结构里也有：队列、优先级队列~

队列：最简单的队列，先进先出，可基于数组实现，也可基于链表实现~

这是最朴素的队列，还衍生出了一些特殊规则的队列~

优先级队列：出队列不是完全按照先进先出了，而是优先级高的先出~

内部基于堆（完全二叉树）来实现的~

阻塞队列在生活中又有很常见的例子~

比如说，做核酸的例子：

做核酸的时候，经常是有好多个队伍在排队，假设现在某个队伍已经没人了，那么做核酸的工作人员不会立即走的，还会等待，等待新的人过来，一直到约定好的时间~

再假设现在的人比较多，张三同学过来做核酸了，他看了看人比较多，也不想在太阳底下晒着，就先在树荫里玩一会手机，等人少一点然后再过去排队~

一、阻塞队列概论

1.1 阻塞队列的概念与作用

阻塞队列实际上还是一种队列，遵循 "先进先出、后进后出" 的原则，它能够保证 "线程安全"~

其主要特点是：

如果队列为空，尝试出队列就会阻塞~
如果队列满，尝试入队列也会阻塞~

阻塞：让线程停下来等一等，本质上就是修改了线程的状态，让线程的 PCB 在内核中暂时不参与调度~

1.2 阻塞队列的应用场景 —— 生产者消费者模型

阻塞队列，其中的一个最重要的场景：生产者消费者模型~

咱们可以例举出一个场景来帮助理解什么叫做生产者消费者模型：

过年的时候包饺子，都是一家人围在一起，一起来包饺子~

包饺子里面有很多道工序，现在就简化一下步骤，假设只有以下步骤：擀饺子皮 —> 包饺子~

当人多的时候来包饺子的时候，会有这两种情况：

每个人都分别自己擀皮，自己包饺子~
有一个人专门负责擀皮，其他人来包饺子~

第一种情况：我们把每一个人想象成一个线程，每个线程都分别完成 "擀皮" 和 "包饺子" 这两样工作~

第二种情况：我们还是把每一个人想象成一个线程，一个线程负责 "擀皮"，其它线程负责 "包饺子"~

这两种情况都是多线程编程典型的解决问题的方式~

但是，其实还是有一点点缺陷的：

第一种情况 的效率比较低，多个线程都在抢同一个资源（擀面杖，正常家庭也就只有一个擀面杖），只有拿到 "擀面杖" 的线程才可以 "擀皮"，其他的线程就可能在摸鱼、等着~

第二种情况 的效率就比较高效，擀皮的人一直都使用擀面杖，其他的人不使用~

而这第二种情况，我们就把它叫做生产者消费者模型！！！

盖帘就是一个阻塞队列，阻塞队列的特点在这上面也是可以体现的~

比如说，极端情况下，负责包饺子的消费者包的太快了，使得负责擀皮的生产者跟不上了，盖帘上面的饺子皮都没有了，所以负责包饺子的消费者就只能阻塞等待，直到擀皮的人擀了一个新的饺子皮以后再取走~

另外一种极端情况，擀皮的人搞得太快了，包饺子的人包的太慢了，一顿操作猛如虎，擀皮的人把盖帘搞满了，所以此时负责擀皮的人只能再等待一会~

像这样的场景，我们把它叫做生产者消费者模型！！！

1.3 生产者消费者模型的好处

使用生产者消费者模型，在工作中是非常频繁的~

优点有很多，其中最为明显的优点有两条：

（一）可以做到更好的 "解耦合"~

耦合：两个模块的关联关系越紧密，就说明耦合程度越高，一边出问题就会导致另一边出现问题，一边出问题就会对另一边有影响~

在写代码的时候，我们追求的都是 "低耦合"，我们都希望，万一某一个模块出现了问题，另外一边还可以照常的工作，不会有太大的影响~

如果服务器A 突然又发了一个数据给服务器C，那么在引入 "生产者消费者模型" 之前，就需要重新调整服务器A所需发送的数据的代码；但是，在引入 "生产者消费者模型" 之后，就不需要重新调整了，只需要服务器C 去队列里取数据就可以了~

（二）使用生产者消费者模型，可以提高整个系统的抗风险能力~

极端情况下，大规模用户访问服务器A(请求服务器)，如果不通过阻塞队列，服务器A 把所有的请求的数据都同步转给服务器B，服务器B(应用服务器)没有太强的抗压能力，就很容易搞挂了！！！

可以类似于三峡大坝的 "削峰填谷" 的功能：在旱季放水，在涝季存水~

二、阻塞队列的实现

2.1 Java标准库里面的实现

首先，在 Java 标准库里面，提供了一个现成的实现~

BlockingQueue 虽然也能够使用 offer、poll 等普通队列的方法，但是仍然建议大家使用 put 来入队列，使用 take 来出队列（这样可以做到阻塞的效果）~

我们可以使用 sleep方法 来模拟生产者生产、消费者消费的频率~

情况一：生产者生产频率 = 消费者消费频率~

package thread;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
public class Demo21 {
public static void main(String[] args) {
//内部基于链表实现 LinkedBlockingDeque
BlockingDeque<Integer> queue = new LinkedBlockingDeque<>();
//创建一个消费者线程
//消费者 每秒消费1个
Thread customer = new Thread(()-> {
//获取队列里面的元素
while (true) {
try {
int value = queue.take();//自动拆箱
System.out.println("消费元素：" + value);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
customer.start();
//创建一个生产者线程
//生产者每秒生产1个
Thread producer = new Thread(() -> {
int n = 0;
while (true) {
System.out.println("生产元素：" + n);
try {
queue.put(n);
n++;
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
}
}

运行结果：

生产者与消费者频率一致~

生产者刚生产好，就立即消费者被消费~

此时消费者步调与生产者一致~

情况二：生产者生产频率 > 消费者消费频率~

使得生产者每 1 秒生产一个，消费者每 2 秒消费一个（代码都和上面一样，只不过把 sleep() 里面的时间修改了，所以就不做过多解释了）

运行结果：

生产者生产快一点，消费者消费慢一点~

阻塞队列满了之后，生产者需要等待消费者消费后才能生产~

此时生产者步调与消费者一致~

情况三：生产者生产频率 < 消费者消费频率~

使得生产者每 2 秒生产一个，消费者每 1 秒消费一个（代码都和上面一样，只不过把 sleep() 里面的时间修改了，所以就不做过多解释了）

运行结果：

生产者生产慢一点，消费者消费快一点~

阻塞队列为空之后，消费者需要等待生产者生产，消费者才能消费~

此时消费者步调与生产者一致~

2.2 自己动手去模拟实现一个阻塞队列

2.2.1 首先实现一个普通队列

在这之前，我们需要模拟一个普通队列~

//基于数组的方式来实现，提供 2 个核心方法：
// 1.put方法 入队列
// 2.take方法 出队列
class MyBlockingQueue {
//假定最大是 1000 个元素，当然也可以设定成 可配置的
private int[] items = new int[1000];
//对首的位置
private int head = 0;
//对尾的位置
private int tail = 0;
//队列的元素个数
private int size = 0;
//入队列
public void put(int value) {
synchronized (this) {
}
if(size == items.length) {
//队列已满，无法插入
return;
}
//队列没满，入队列
items[tail] = value;
tail++;
if (tail == items.length){
//判断 tail 是否到达末尾
//如果 tail 到达末尾，就需要从头开始
tail = 0;
}
//往上数五行代码，可以直接换成 tail = tail % items.length;
//不过带来了一些问题
//1.可读性不太好~ 写 if 做判断，一看就明白了；写 % 运算，也许其他人看不懂
//2.当使用 % 运算的时候，只有说 % 后面的操作数是 2^n 的时候，% 才能有一个比较高效的计算过程 (被编译器优化成与运算)
//
如果是随意给的运算，大概率是不行了
//个人建议 if() 版本
size++;
}
//出队列
public Integer take() { //int 不可以返回 null，Integer 可以返回 null
if (size == 0) {
//队列为空，无法出队列
return null;
}
//队列不为空，则 取出队首元素
int ret = items[head];
head++;
if (head == items.length) {
head = 0;
}
size--;
return ret;
}
}

接着，我们可以在普通队列的基础上进行改进~

线程安全 —— 加锁、volatile
阻塞 —— wait、notify

然后就自己模拟实现了一个阻塞队列啦~

2.2.2 接着需要解决线程安全问题

单例模式之所以使用两个 if 的原因，是因为单例模式只是在初始化阶段有线程安全问题，一旦初始化好了，就线程安全了，所以需要使用外层条件来决定当前是否要加锁~

而当下的阻塞队列，是自始至终都有线程安全问题的，这个锁就得要始终加上~

需要注意的是，多线程这里，要不要加锁，具体锁加在哪里，是没有规律的，只能具体问题具体分析（千万不要无脑加锁）~

当然，这个阻塞队列到处都是修改和读操作，就可以无脑加锁了~

当然，队列的元素个数时不时的在读，而且在修改，加上 volatile 可能会更好~

//基于数组的方式来实现，提供 2 个核心方法：
// 1.put方法 入队列
// 2.take方法 出队列
class MyBlockingQueue {
//假定最大是 1000 个元素，当然也可以设定成 可配置的
private int[] items = new int[1000];
//对首的位置
private int head = 0;
//对尾的位置
private int tail = 0;
//队列的元素个数
volatile private int size = 0;
//入队列
public void put(int value) {
synchronized (this) {
if(size == items.length) {
//队列已满，无法插入
return;
}
//队列没满，入队列
items[tail] = value;
tail++;
if (tail == items.length){
//判断 tail 是否到达末尾
//如果 tail 到达末尾，就需要从头开始
tail = 0;
}
//往上数五行代码，可以直接换成 tail = tail % items.length;
//不过带来了一些问题
//1.可读性不太好~ 写 if 做判断，一看就明白了；写 % 运算，也许其他人看不懂
//2.当使用 % 运算的时候，只有说 % 后面的操作数是 2^n 的时候，% 才能有一个比较高效的计算过程 (被编译器优化成与运算)
//
如果是随意给的运算，大概率是不行了
//个人建议 if() 版本
size++;
}
}
//出队列
public Integer take() { //int 不可以返回 null，Integer 可以返回 null
int ret = 0;
synchronized (this) {
if (size == 0) {
//队列为空，无法出队列
return null;
}
//队列不为空，则 取出队首元素
ret = items[head];
head++;
if (head == items.length) {
head = 0;
}
size--;
}
return ret;
}
}

2.2.3 最后来实现阻塞效果

阻塞有两种情况的：

队列为空，要阻塞，当队列不空的时候就唤醒
队列为满，要阻塞。当队列不满的时候就唤醒

所以，我们可以把 put 和 take 方法改成如下形式：

注意：

虽然按照上述的代码，发现唤醒的时候，一定是有元素插入成功了，所以条件不成立，等待确实是要结束了~

但是，更稳妥的办法，是在唤醒之后，再判断一次（万一条件又成立了呢）！！！

所以，可以把 if 条件句改成 while 循环语句：

我们等待之前，判断一次；唤醒之后，再确认一次~

所以，最终的代码就出来了：

package thread;
//自己来模拟实现一个阻塞队列
//基于数组的方式来实现，提供 2 个核心方法：
// 1. put方法 入队列
// 2.take方法 出队列
class MyBlockingQueue {
//假定最大是 1000 个元素，当然也可以设定成 可配置的
private int[] items = new int[1000];
//对首的位置
private int head = 0;
//对尾的位置
private int tail = 0;
//队列的元素个数
volatile private int size = 0;
//入队列
public void put(int value) throws InterruptedException {
synchronized (this) {
while(size == items.length) {
//队列已满，无法插入
this.wait();
}
//队列没满，入队列
items[tail] = value;
tail++;
if (tail == items.length){
//判断 tail 是否到达末尾
//如果 tail 到达末尾，就需要从头开始
tail = 0;
}
//往上数五行代码，可以直接换成 tail = tail % items.length;
//不过带来了一些问题
//1.可读性不太好~ 写 if 做判断，一看就明白了；写 % 运算，也许其他人看不懂
//2.当使用 % 运算的时候，只有说 % 后面的操作数是 2^n 的时候，% 才能有一个比较高效的计算过程 (被编译器优化成与运算)
//
如果是随意给的运算，大概率是不行了
//个人建议 if() 版本
size++;
//插入元素成功，说明队列不空，就要唤醒
this.notify();
}
}
//出队列
public Integer take() throws InterruptedException { //int 不可以返回 null，Integer 可以返回 null
int ret = 0;
synchronized (this) {
while(size == 0) {
//队列为空，就等待
this.wait();
}
//队列不为空，则 取出队首元素
ret = items[head];
head++;
if (head == items.length) {
head = 0;
}
size--;
//当取走一个元素成功，说明队列不满，就要唤醒
this.notify();
}
return ret;
}
}
public class Demo22 {
public static void main(String[] args) throws InterruptedException {
MyBlockingQueue queue = new MyBlockingQueue();
Thread customer = new Thread(() -> {
while (true) {
int value = 0;
try {
value = queue.take();
System.out.println("消费：" + value);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
customer.start();
Thread producer = new Thread(() -> {
int value = 0;
while (true){
try {
queue.put(value);
System.out.println("生产：" + value);
value++;
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
}
}

运行结果：