认识阻塞队列

1.1 阻塞队列是什么

阻塞队列是一种特殊的队列&＃xff0c;也遵守 “先进先出” 的原则。

1&＃xff09;是一种线程安全的数据结构&＃xff1b;
2&＃xff09;带有阻塞功能&＃xff1a;

• 当队列满的时候, 继续入队列就会阻塞&＃xff0c;直到有其他线程从队列中取走元素&＃xff1b;
• 当队列空的时候, 继续出队列也会阻塞&＃xff0c;直到有其他线程往队列中插入元素。

阻塞队列的一个典型应用场景就是 “生产者消费者模型”&＃xff0c;这是一种非常典型的开发模型&＃xff01;

1.2 生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。
生产者和消费者彼此之间不直接通讯&＃xff0c;而通过阻塞队列来进行通讯&＃xff0c;所以生产者生产完数据之后不用等待消费者处理&＃xff0c;直接扔给阻塞队列&＃xff0c;消费者不找生产者要数据&＃xff0c;而是直接从阻塞队列里取。

1&＃xff09;阻塞队列能使生产者和消费者之间解耦合&＃xff01;

比如过年一家人一起包饺子&＃xff0c;一般都是有明确分工&＃xff0c;比如一个人负责擀饺子皮&＃xff0c;其他人负责包&＃xff0c;擀饺子皮的人就是 “生产者”&＃xff0c;包饺子的人就是 “消费者”。
擀饺子皮的人不关心包饺子的人是谁 (能包就行&＃xff0c;无论是手工包&＃xff0c;借助工具&＃xff0c;还是机器包)&＃xff0c;包饺子的人也不关心擀饺子皮的人是谁 (有饺子皮就行&＃xff0c;无论是用擀面杖擀的&＃xff0c;还是拿罐头瓶擀&＃xff0c;还是直接从超市买的)。

2&＃xff09;阻塞队列就相当于一个缓冲区&＃xff0c;平衡了生产者和消费者的处理能力&＃xff01;削峰填谷&＃xff01;

比如在 “秒杀” 场景下&＃xff0c;服务器同一时刻可能会收到大量的支付请求。如果直接处理这些支付请求&＃xff0c;服务器可能扛不住 (每个支付请求的处理都需要比较复杂的流程)。这个时候就可以把这些请求都放到一个阻塞队列中&＃xff0c;然后再由消费者线程慢慢的来处理每个支付请求。这样做可以有效进行 “削峰”&＃xff0c;防止服务器被突然到来的一波请求直接冲垮。
比如三峡大坝&＃xff1a;

2.1 使用

在 Java 标准库中内置了阻塞队列。如果我们需要在一些程序中使用阻塞队列&＃xff0c;直接使用标准库中的即可&＃xff1a;

• BlockingQueue 是一个接口&＃xff0c;真正实现的类是 LinkedBlockingQueue、ArrayBlockingQueue等…
• put 方法用于阻塞式的入队列&＃xff0c;take 用于阻塞式的出队列
• BlockingQueue 也有 offer、poll、peek 等方法&＃xff0c;但是这些方法不带有阻塞特性

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class Demo {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<Integer> blockingQueue &＃61; new LinkedBlockingQueue<>(100);
// 带有阻塞功能的入队列
blockingQueue.put(1);
blockingQueue.put(2);
blockingQueue.put(3);
Integer ret &＃61; blockingQueue.take();
System.out.println(ret);
ret &＃61; blockingQueue.take();
System.out.println(ret);
ret &＃61; blockingQueue.take();
System.out.println(ret);
ret &＃61; blockingQueue.take();
System.out.println(ret);
}
}


注意&＃xff1a; 阻塞队列没有提供"带有阻塞的"取队首元素的功能&＃xff0c;所以需要先take取出判断&＃xff0c;不符合再重新put进队列&＃xff01;

2.2 生产者消费者模型实现

public static void main(String[] args) throws InterruptedException {
BlockingQueue<Integer> blockingQueue &＃61; new LinkedBlockingQueue<Integer>();
Thread customer &＃61; new Thread(() -> {
while (true) {
try {
int value &＃61; blockingQueue.take();
System.out.println("消费元素: " &＃43; value);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "消费者");
customer.start();
Thread producer &＃61; new Thread(() -> {
Random random &＃61; new Random();
while (true) {
try {
int num &＃61; random.nextInt(1000);
System.out.println("生产元素: " &＃43; num);
blockingQueue.put(num);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "生产者");
producer.start();
customer.join();
producer.join();
}

三、实现阻塞队列

3.1 循环队列

我们来基于数据实现&＃xff1a;循环队列&＃xff01;

怎样区别"空"还是"满"&＃xff1f;
1&＃xff09;浪费一个数组空间&＃xff1b;2&＃xff09;添加计数器

循环队列更多内容~~请点击

3.2 实现的细节

保证线程安全&＃xff0c;关键就是加锁&＃xff01;
实现阻塞&＃xff1a;
1.如果队列为空&＃xff0c;出队列就阻塞&＃xff1b;
2.如果队列为满&＃xff0c;入队列就阻塞。

即使没有t3这一手&＃xff0c;还有interrupt&＃xff0c;也是有影响~~
所以&＃xff0c;使用while是稳妥的写法&＃xff01;&＃xff01;&＃xff01;

3.3 代码

// 名字还是不要和标准库的混淆
class MyBlockingQueue {
private int[] items &＃61; new int[1000];
private volatile int head &＃61; 0;
private volatile int tail &＃61; 0;
private volatile int size &＃61; 0;
// 入队列
public void put(int elem) throws InterruptedException {
synchronized (this) {
// 判定队列是否满了, 满了则不能插入.
while (size >&＃61; items.length) {
this.wait();
}
// 进行插入操作, 把 elem 放到 items 里, 放到 tail 指向的位置.
items[tail] &＃61; elem;
tail&＃43;&＃43;;
if (tail >&＃61; items.length) {
tail &＃61; 0;
}
size&＃43;&＃43;;
this.notify();
}
}
// 出队列, 返回删除的元素内容
public Integer take() throws InterruptedException {
synchronized (this) {
// 判定队列是否空, 如果空了, 则不能出队列
while (size &＃61;&＃61; 0) {
this.wait();
}
// 进行取元素操作.
int ret &＃61; items[head];
head&＃43;&＃43;;
if (head >&＃61; items.length) {
head &＃61; 0;
}
size--;
this.notify();
return ret;
}
}
}
public class Demo {
public static void main(String[] args) {
MyBlockingQueue queue &＃61; new MyBlockingQueue();
Thread producer &＃61; new Thread(() -> {
int n &＃61; 1;
while (true) {
try {
queue.put(n);
System.out.println("生产元素 " &＃43; n);
n&＃43;&＃43;;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread customer &＃61; new Thread(() -> {
while (true) {
try {
int n &＃61; queue.take();
System.out.println("消费元素 " &＃43; n);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
customer.start();
}
}