了解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;
}
@Override
public 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;
}
@Override
public 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;
}
@Override
public void run() {
try {
while (true) {
Runnable task = taskQueue.take();
task.run();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}
参考链接
- Baeldung: Java BlockingQueue
