【阻塞队列】阻塞队列的模拟实现及在生产者和消费者模型上的应用

📄前言

本文是对阻塞队列的应用场景的介绍，对阻塞队列的作用以及具体实现的讨论。

---

一. 阻塞队列初了解

🍆1. 什么是阻塞队列？

阻塞队列是一种带有阻塞功能的“先进先出”线性表。即在一个带有最大容量的队列中，在某时刻队列容量已满时继续入队 或 队列为空时继续出队，就会进入阻塞等待状态，直到队列变为 未满或非空 便解除阻塞状态，继续入队或出队。

🍅2. 为什么使用阻塞队列？

若存在以下简易的分布式系统：

上述分布式系统虽然能完成客户端与服务器端的交互需求，但可能存在以下问题：

1. 在正常情况下，用户可以通过客户端想服务器发起请求并获取相应的服务，但假如在某刻服务器A突然出现了故障，与服务器A直接通信的服务器B也可能因此出现故障，导致整个服务瘫痪。
2. 若未来想增加 更多的服务器 来处理服务器A发起的请求，则需求对 服务器A 的接口 进行一定的改动，付出一定的时间和人力成本。
3. 当某个时刻，很多的客户端同时向 服务器A 发起请求，作为与用户直接交互的服务器，服务器A具备承载这些并发量的能力，但服务器集群中负责其他功能的服务器接收请求的承载能力可能较弱，此时可能造成其他服务器的崩溃。

造成上述现象的原因可以归结为以下两点：

1. 模块间的耦合性较高（例如问题1和2）
2. 承载能力较弱的模块不具备抗冲击能力。（例如问题3）

---

上述的解决方法是在服务器之间加入一个阻塞队列，利用生产者和消费者模型解决以上问题。
什么是生产者消费者模型呢？（如下图）

当服务器A接收来自客户端的请求时，不把请求直接发给服务器B，而是将请求数据加入到队列中，服务器B通过队列接收请求并把请求除了的结果返回给A。

---

当上述分布式系统引入阻塞队列后工作模式如下图所示：

引入阻塞队列的好处：

1. 解耦合。当服务器A或服务器B出现问题时，就不会对其他服务器造成直接的影响；当需要添加新的服务器来处理这些请求时，新的服务器也同样只需从队列中取数据，无需对原有服务器的接口（代码）进行任何的改动。
2. “削峰填谷”。当服务器A 瞬间接收客户端发来的大量请求时，由于服务器B处理请求的速度较慢，剩余的请求会在阻塞队列里面堆积，虽然客户端获取服务的时间相对增加了，但一定程度上缓解了其他承受并发量能力较弱的服务器的压力。

🥦3. Java标准库中阻塞队列的实现

BlockingQueue的主要方法：

方法演示如下：（使用普通入队方法入队4次，再使用带有阻塞的出队方法出队4次）

public static void main(String[] args) throws InterruptedException {BlockingQueue<Integer> q = new ArrayBlockingQueue<>(3);System.out.println("数据 5 入队状态: " + q.offer(5));System.out.println("数据 6 入队状态: " + q.offer(6));System.out.println("数据 7 入队状态: " + q.offer(7));System.out.println("数据 8 入队状态: " + q.offer(8));System.out.print("队列中的数据: ");System.out.println(q);System.out.println("数据出队: ");for (int i = 0; i < 4; i++) {System.out.print(q.take() + " ");}System.out.println("程序结束 !");
}

可以发现，当调用 take()方法取出队列元素时，因为队列最终为空，程序进入了阻塞状态，没有打印“程序结束”。

---

二. 阻塞队列的模拟实现

🍚1. 实现普通队列

阻塞队列的关键方法是两个带有阻塞功能的 put() 和 take()方法，而这两个方法是在原有出入队方法上使用 Object类 带有wait()方法 和 notify() 方法让线程进入等待状态 或 唤醒线程。
因此，我们可以先把基础的队列进行实现，随后在原有基础上进行修改。队列可以使用数组（环形队列）或链表两种方式实现，这里我采用数组的方式实现队列。（由于队列的实现方法较为常见，这里直接给出实现代码）

class MyBlockingQueue<E> {private Object[] elem;private int defaultCapacity = 11;	// 阻塞队列默认容量private int front;	// 记录队头元素位置private int rear;	// 记录队尾元素位置private int size;   // 用于记录当前队列元素的实际个数public MyBlockingQueue(){this.elem = new Object[defaultCapacity + 1];}public MyBlockingQueue(int capacity) {defaultCapacity = capacity;this.elem = new Object[defaultCapacity + 1];}public boolean offer(E val) {// 判断队列是否已满if (size == defaultCapacity) {return false;}elem[rear] = val;size++;// 如果 rear自增 到达数组末尾，使 rear 重新到数组的头部rear = (rear + 1) % (defaultCapacity + 1);return true;}public E poll() {// 判断队列是否为空if (front == rear) {return null;}Object ret = elem[front];size--;// 如果 front 自增 到达数组末尾，使 front 重新到数组的头部front = (front + 1) % (defaultCapacity + 1);return (E)ret;}
}

🍥2. 实现队列的阻塞功能

当阻塞队列容量已满时，调用 put() 方法会进入阻塞状态，因此在原先 offer()方法判断的基础上，我们需要使用 wait()方法 让线程进入阻塞等待状态，考虑到可能有多个线程同时调用 put()方法，可能会引起线程安全问题，因此我们应在 if()判断条件和整个修改操作上 加锁(或者直接在方法上加锁)。（代码如下）

public void put (E value) throws InterruptedException {// 判断队列是否已满synchronized (this) {if (size == defaultCapacity) {// 队列进入阻塞状态， 直到有元素出队时 解除阻塞this.wait();}queue[rear] = value;size++;// 如果 rear自增 到达数组末尾，使 rear 重新到数组的头部rear = (rear + 1) % (defaultCapacity + 1);}
}

当队列为空时，调用 take() 方法会使线程进入阻塞状态，同理若判空条件成立，我们需要调用 wait() 方法使线程进入阻塞，为防止多个线程在队列即将为空时同时调用 take() 方法引发线程安全问题，我们需要在 if()判断语句 和 整个修改操作 进行加锁操作(或者直接在方法上加锁)。（代码如下）

public E take() throws InterruptedException {// 判断队列是否为空synchronized (this) {if (rear == front) {// 队列进入阻塞状态，直到有新的元素入队时 解除阻塞this.wait();}Object ret = queue[front];// 如果 front 自增 到达数组末尾，使 front 重新到数组的头部front = (front + 1) % (defaultCapacity + 1);size--;return (E)ret;}
}

🧊3. 解除阻塞状态

什么情况下队列会接触阻塞状态呢？

1. 当队满时，某个线程从阻塞队列取出一个元素，即执行完出队操作后，需要使用 notify()方法 唤醒因执行 put()方法而阻塞的线程。
2. 当队空时，某个线程向队列新增一个元素，即执行完入队操作后，需要使用 notify()方法唤醒因执行 take()方法而阻塞的线程。

对 put()方法和take()方法 修改后代码如下：

public void put (E value) throws InterruptedException {// 判断队列是否已满synchronized (this) {if (size == defaultCapacity) {// 队列进入阻塞状态， 直到有元素出队时 解除阻塞this.wait();}queue[rear] = value;size++;// 如果 rear自增 到达数组末尾，使 rear 重新到数组的头部rear = (rear + 1) % (defaultCapacity + 1);// 此处的 notify 用来唤醒 队列为空时的 waitthis.notify();}
}public E take() throws InterruptedException {// 判断队列是否为空synchronized (this) {if (rear == front) {// 队列进入阻塞状态，直到有新的元素入队时 解除阻塞this.wait();}Object ret = queue[front];// 如果 front 自增 到达数组末尾，使 front 重新到数组的头部front = (front + 1) % (defaultCapacity + 1);size--;// 此处的 notify 用来唤醒 队列为满时的 waitthis.notify();return (E)ret;}
}

---

三. 使用模拟的阻塞队列验证生产者和消费者模型

为了方便看到效果，我们假设阻塞队列的容量为2，并将生产与消费的数据进行打印。
当生产者与消费者处理数据的频率一样，且生产速率为 次/1s、消费速率为 次/1s 时，程序的生产与消费数据应轮流打印：（模拟代码和程序运行结果如下）

public static void main(String[] args) {MyBlockingQueue<Integer> myBlockingQueue = new MyBlockingQueue<>(2);// 生产者Thread producer = new Thread(() -> {for (int i = 0; i < 5; i++) {try {myBlockingQueue.put(i);System.out.println("生产了: " + i);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});// 消费者Thread consumer = new Thread(() -> {for (int i = 0; i < 5; i++) {try {int ret = myBlockingQueue.take();System.out.println("消费了: " + ret);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}

当生产速率 > 消费速率，且生产速率为 次/1s、消费速率为 次/2s 时：可以预估到，当经过5s后程序会因队满进入阻塞状态，且后续每消费一次伴随着一次生产，为方便观察阻塞情况，我们可以在方法实现的地方加上阻塞日志的提示（模拟代码和程序运行结果如下）

public static void main(String[] args) {MyBlockingQueue<Integer> myBlockingQueue = new MyBlockingQueue<>(2);Thread producer = new Thread(() -> {for (int i = 0; i < 10; i++) {try {myBlockingQueue.put(i);System.out.println("生产了: " + i);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});Thread consumer = new Thread(() -> {for (int i = 0; i < 10; i++) {try {int ret = myBlockingQueue.take();System.out.println("消费了: " + ret);Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}

当生产速率 < 消费速率，且生产速率为 次/2s、消费速率为 次/1s 时：可以预估到，当经过2s后程序会因队满进入阻塞状态，且后续每生产一次伴随着一次消费，为方便观察阻塞情况，我们可以在方法实现的地方加上阻塞日志的提示（模拟代码和程序运行结果如下）

public static void main(String[] args) {MyBlockingQueue<Integer> myBlockingQueue = new MyBlockingQueue<>(2);Thread producer = new Thread(() -> {for (int i = 0; i < 10; i++) {try {myBlockingQueue.put(i);System.out.println("生产了: " + i);Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});Thread consumer = new Thread(() -> {for (int i = 0; i < 10; i++) {try {int ret = myBlockingQueue.take();System.out.println("消费了: " + ret);Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});producer.start();consumer.start();
}

---

以上就是本篇文章的全部内容了，如果这篇文章对你有些许帮助，你的点赞、收藏和评论就是对我最大的支持。
另外，文章可能存在许多不足之处，也希望你可以给我一点小小的建议，我会努力检查并改进。