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

了解Java的LinkedBlockingQueue

news 来源：原创 2024/7/6 7:44:18

了解Java的LinkedBlockingQueue

LinkedBlockingQueue是一个基于链接节点的有界阻塞队列。它实现了BlockingQueue接口，可以在多线程环境中安全地进行插入、移除和检查操作。LinkedBlockingQueue的容量可以在创建时指定，如果未指定，则默认容量为Integer的最大值。

线程安全

LinkedBlockingQueue通过以下机制实现线程安全：

独占锁（ReentrantLock）：队列内部使用两把不同的独占锁来管理入队和出队操作。入队操作和出队操作分别使用不同的锁，从而实现了入队和出队的并行操作，提高了性能。
条件变量（notEmpty和notFull）：
- notEmpty：用于等待队列不为空的条件。当消费者线程发现队列为空时，会在notEmpty上等待，直到有元素被生产者放入队列。
- notFull：用于等待队列未满的条件。当生产者线程发现队列已满时，会在notFull上等待，直到有空间被消费者释放。
节点链接结构：LinkedBlockingQueue使用链表节点来存储数据，每个节点包含一个数据元素和指向下一个节点的引用。入队操作会在链表的尾部插入新节点，出队操作会从链表的头部移除节点。
容量限制：LinkedBlockingQueue可以通过构造函数指定容量，如果未指定则默认容量为Integer.MAX_VALUE。在插入元素时，如果队列已满，插入操作会被阻塞，直到有空间可用；在移除元素时，如果队列为空，移除操作会被阻塞，直到有元素可用。

通过上述机制，LinkedBlockingQueue能够在多线程环境中保证线程安全，同时在性能和资源利用率之间取得平衡。

使用方法

创建队列

可以通过指定容量来创建LinkedBlockingQueue：

LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);

如果不指定容量，队列的默认容量为Integer.MAX_VALUE：

LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

插入元素

LinkedBlockingQueue提供了多种插入元素的方法：

put(E e)：如果队列已满，则等待直到队列不再满。
offer(E e)：如果队列已满，则返回false。
offer(E e, long timeout, TimeUnit unit)：在指定的时间内等待可用空间，如果超时则返回false。

queue.put(1);
boolean success = queue.offer(2);
boolean successWithTimeout = queue.offer(3, 2, TimeUnit.SECONDS);

移除元素

LinkedBlockingQueue提供了多种移除元素的方法：

take()：如果队列为空，则等待直到有元素可用。
poll()：如果队列为空，则返回null。
poll(long timeout, TimeUnit unit)：在指定的时间内等待元素可用，如果超时则返回null。

Integer item = queue.take();
Integer itemOrNull = queue.poll();
Integer itemOrNullWithTimeout = queue.poll(2, TimeUnit.SECONDS);

检查元素

LinkedBlockingQueue提供了检查元素的方法：

peek()：返回队列头部的元素，但不移除它，如果队列为空，则返回null。

Integer head = queue.peek();

应用场景

生产者-消费者模式

LinkedBlockingQueue非常适用于生产者-消费者模式。在这种模式中，生产者线程负责生产数据并将其放入队列中，消费者线程从队列中取出数据进行处理。LinkedBlockingQueue的阻塞特性可以有效地协调生产者和消费者的速度，避免数据丢失和资源浪费。

示例代码：

class Producer implements Runnable {private final LinkedBlockingQueue<Integer> queue;public Producer(LinkedBlockingQueue<Integer> queue) {this.queue = queue;}@Overridepublic void run() {try {while (true) {int item = produce();queue.put(item);}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}private int produce() {// 生产数据的逻辑return new Random().nextInt();}
}class Consumer implements Runnable {private final LinkedBlockingQueue<Integer> queue;public Consumer(LinkedBlockingQueue<Integer> queue) {this.queue = queue;}@Overridepublic void run() {try {while (true) {int item = queue.take();consume(item);}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}private void consume(int item) {// 消费数据的逻辑}
}public class Main {public static void main(String[] args) {LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);Producer producer = new Producer(queue);Consumer consumer = new Consumer(queue);new Thread(producer).start();new Thread(consumer).start();}
}

任务调度

在任务调度系统中，可以使用LinkedBlockingQueue来管理任务队列。调度器线程将任务添加到队列中，工作线程从队列中获取任务并执行。这样可以实现任务的均衡分配和并发处理，提高系统的响应速度和处理能力。

示例代码：

class TaskScheduler {private final LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();private final List<Thread> workers = new ArrayList<>();public TaskScheduler(int numberOfWorkers) {for (int i = 0; i < numberOfWorkers; i++) {workers.add(new Thread(new Worker(taskQueue)));}for (Thread worker : workers) {worker.start();}}public void schedule(Runnable task) throws InterruptedException {taskQueue.put(task);}private static class Worker implements Runnable {private final LinkedBlockingQueue<Runnable> taskQueue;public Worker(LinkedBlockingQueue<Runnable> taskQueue) {this.taskQueue = taskQueue;}@Overridepublic void run() {try {while (true) {Runnable task = taskQueue.take();task.run();}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}
}