【JavaEE】阻塞队列，代码模拟实现阻塞队列并应用于生产者消费者模型

了解阻塞队列

阻塞队列是Java中的一种线程安全的队列，通常用于实现生产者-消费者模式。它不仅可以存储数据，还提供了当队列为空或已满时线程的阻塞能力。阻塞队列在多线程环境中非常有用，可以有效地解决线程间的通信和协调问题。

在Java中，阻塞队列的主要实现有：

ArrayBlockingQueue：基于数组结构的阻塞队列，容量有限，按FIFO（先进先出）原则操作。
LinkedBlockingQueue：基于链表结构的阻塞队列，容量可选择无限或指定，具有较高的并发性能。
PriorityBlockingQueue：支持优先级排序的阻塞队列，元素按照自然顺序或提供的比较器排序。
DelayQueue：一个支持延迟排队的阻塞队列，只有在元素到达预定时间后才能被 consumed。

阻塞队列提供了一些重要的方法，如put()和take()，用于插入和获取元素。这些方法在队列满或空时会导致调用线程阻塞，从而有效控制线程的执行流程。

总之，阻塞队列是处理多线程编程中任务调度及数据共享的一个重要工具，能提高代码的可读性和可靠性。

代码模拟实现阻塞队列

模拟实现

下面是一个简单的Java实现的阻塞队列示例。这个实现使用了wait()和notifyAll()来处理线程的阻塞和唤醒。我们的自定义阻塞队列类 MyBlockingQueue 支持基本的 put 和 take 方法。

import java.util.LinkedList;public class MyBlockingQueue<T> {private final LinkedList<T> queue;private final int capacity;public MyBlockingQueue(int capacity) {this.queue = new LinkedList<>();this.capacity = capacity;}// 添加元素到队列public synchronized void put(T item) throws InterruptedException {while (queue.size() == capacity) {wait(); // 队列满时，等待}queue.add(item);notifyAll(); // 唤醒其他线程}// 从队列中获取元素public synchronized T take() throws InterruptedException {while (queue.isEmpty()) {wait(); // 队列为空时，等待}T item = queue.removeFirst();notifyAll(); // 唤醒其他线程return item;}// 查看队列大小public synchronized int size() {return queue.size();}// 检查队列是否为空public synchronized boolean isEmpty() {return queue.isEmpty();}
}

使用示例

下面是一个使用上述 MyBlockingQueue 的示例，包括生产者和消费者的实现：

public class BlockingQueueExample {public static void main(String[] args) {MyBlockingQueue<Integer> queue = new MyBlockingQueue<>(5); // 创建容量为5的阻塞队列// 生产者线程Thread producer = new Thread(() -> {try {for (int i = 0; i < 10; i++) {queue.put(i);System.out.println("Produced: " + i);Thread.sleep(100); // 模拟生产时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});// 消费者线程Thread consumer = new Thread(() -> {try {for (int i = 0; i < 10; i++) {Integer value = queue.take();System.out.println("Consumed: " + value);Thread.sleep(150); // 模拟消费时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});producer.start();consumer.start();}
}

总结

上面的代码展示了如何实现一个简单的阻塞队列以及生产者-消费者模式的基本用法。你可以根据需要扩展这个基础实现，例如添加更多的功能和错误处理。

代码解析

下面是对上述代码实现的详细解析，包括 MyBlockingQueue 类和生产者-消费者示例的工作原理。

1. `MyBlockingQueue` 类

这个类实现了一个基于链表的阻塞队列，包括两个主要的方法 put 和 take，以及一些辅助方法。

属性

queue: 使用 LinkedList 来存储队列中的元素。
capacity: 指定队列的最大容量。

方法

构造方法:

public MyBlockingQueue(int capacity) {this.queue = new LinkedList<>();this.capacity = capacity;
}

这里我们初始化一个空的链表和指定的容量。

put(T item) 方法:

public synchronized void put(T item) throws InterruptedException {while (queue.size() == capacity) {wait(); // 队列满时，等待}queue.add(item);notifyAll(); // 唤醒其他线程
}

使用 synchronized 关键字保证了方法的线程安全性。

如果队列已满，调用 wait() 方法使当前线程等待，直到有空间可以添加新元素。

当队列有空间时，添加元素，然后调用 notifyAll() 方法唤醒所有在调用 wait() 的线程（可能是消费者线程）。

take() 方法:

public synchronized T take() throws InterruptedException {while (queue.isEmpty()) {wait(); // 队列为空时，等待}T item = queue.removeFirst();notifyAll(); // 唤醒其他线程return item;
}

同样地，是一个同步方法。

如果队列为空，调用 wait() 使当前线程等待，直到有元素可以取出。

当有元素可取时，移除并返回队列的第一个元素，并调用 notifyAll() 以唤醒其他线程（如生产者线程）。

辅助方法:

size(): 返回当前队列的大小。
isEmpty(): 检查队列是否为空。

2. 生产者-消费者示例

在 BlockingQueueExample 类中，我们创建了一个阻塞队列，并定义了生产者和消费者线程。

public class BlockingQueueExample {public static void main(String[] args) {MyBlockingQueue<Integer> queue = new MyBlockingQueue<>(5); // 创建容量为5的阻塞队列

这里我们创建了一个最大容量为5的阻塞队列实例。

生产者线程

Thread producer = new Thread(() -> {try {for (int i = 0; i < 10; i++) {queue.put(i);System.out.println("Produced: " + i);Thread.sleep(100); // 模拟生产时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}
});

生产者线程循环10次，生成数据（0到9）。

每次生成一个数据调用 put 方法加入队列。

如果队列已满，生产者会被阻塞，直到有空间。

消费者线程

Thread consumer = new Thread(() -> {try {for (int i = 0; i < 10; i++) {Integer value = queue.take();System.out.println("Consumed: " + value);Thread.sleep(150); // 模拟消费时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}
});