深入浅出BlockingQueue（三）-育师

ArrayBlockingQueue 与 LinkedBlockingQueue 的比较

相同点：ArrayBlockingQueue 和 LinkedBlockingQueue 都是通过 Condition 通知机制来实现可阻塞的插入和删除。

不同点：

ArrayBlockingQueue 基于数组实现，而 LinkedBlockingQueue 基于链表实现；
ArrayBlockingQueue 使用一个单独的 ReentrantLock 来控制对队列的访问，而 LinkedBlockingQueue 使用两个锁（putLock 和 takeLock），一个用于放入操作，另一个用于取出操作。锁分离，很适合生产和消费频率差不多的场景，这样生产和消费互不干涉的执行，能达到不错的效率。

PriorityBlockingQueue

PriorityBlockingQueue 是一个具有优先级排序特性的无界阻塞队列。元素在队列中的排序遵循自然排序或者通过提供的比较器进行定制排序。可以通过实现 Comparable 接口来定义自然排序。

当需要根据优先级来执行任务时，PriorityBlockingQueue 会非常有用。下面的代码演示了如何使用 PriorityBlockingQueue 来管理具有不同优先级的任务。

class Task implements Comparable<Task> { private int priority; private String name; public Task(int priority, String name) { this.priority = priority; this.name = name; } @Override public int compareTo(Task other) { return Integer.compare(other.priority, this.priority); // higher values have higher priority } public String getName() { return name; } } public class PriorityBlockingQueueDemo { public static void main(String[] args) throws InterruptedException { PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(); queue.put(new Task(1, "Low priority task")); queue.put(new Task(50, "High priority task")); queue.put(new Task(10, "Medium priority task")); while (!queue.isEmpty()) { System.out.println(queue.take().getName()); } } }

以上代码创建了一个优先级阻塞队列，并添加了三个具有不同优先级的任务。它们会按优先级从高到低的顺序被取出并打印。代码运行结果如下：

SynchronousQueue

SynchronousQueue 是一个非常特殊的阻塞队列，它不存储任何元素。每一个插入操作必须等待另一个线程的移除操作，反之亦然。因此，SynchronousQueue 的内部实际上是空的，但它允许一个线程向另一个线程逐个传输元素。

SynchronousQueue 允许线程直接将元素交付给另一个线程。因此，如果一个线程尝试插入一个元素，并且有另一个线程尝试移除一个元素，则插入和移除操作将同时成功。

如果想让一个线程将确切的信息直接发送给另一个线程的情况下，可以使用 SynchronousQueue。下面的代码展示了如何使用 SynchronousQueue 进行线程间的通信：

public class SynchronousQueueDemo { public static void main(String[] args) { SynchronousQueue<String> queue = new SynchronousQueue<>(); // Producer Thread new Thread(() -> { try { String event = "SYNCHRONOUS_EVENT"; System.out.println("Putting: " + event); queue.put(event); System.out.println("Put successfully: " + event); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).start(); // Consumer Thread new Thread(() -> { try { String event = queue.take(); System.out.println("Taken: " + event); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).start(); } }

以上代码创建了一个 SynchronousQueue，并在一个线程中插入了一个元素，另一个线程中移除了这个元素。程序运行结果如下：

LinkedTransferQueue

LinkedTransferQueue 是一个基于链表结构的无界传输队列，实现了 TransferQueue 接口，它提供了一种强大的线程间交流机制。它的功能与其他阻塞队列类似，但还包括“转移”语义：允许一个元素直接从生产者传输给消费者，如果消费者已经在等待。如果没有等待的消费者，元素将入队。

常用方法有两个：

transfer(E e)，将元素转移到等待的消费者，如果不存在等待的消费者，则元素会入队并阻塞直到该元素被消费。
tryTransfer(E e)，尝试立即转移元素，如果有消费者正在等待，则传输成功；否则，返回 false。

如果想要更紧密地控制生产者和消费者之间的交互，可以使用 LinkedTransferQueue。

public class LinkedTransferQueueDemo { public static void main(String[] args) throws InterruptedException { LinkedTransferQueue<String> queue = new LinkedTransferQueue<>(); // Consumer Thread new Thread(() -> { try { System.out.println("消费者正在等待获取元素..."); String element = queue.take(); System.out.println("消费者收到: " + element); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }).start(); // Let consumer thread start first TimeUnit.SECONDS.sleep(1); // Producer Thread System.out.println("生产者正在传输元素"); queue.transfer("Hello, World!"); System.out.println("生产者已转移元素"); } }

消费者线程首先启动并等待接收元素。生产者线程调用 transfer 方法将元素直接传输给消费者。

程序运行结果如下：

LinkedBlockingDeque

LinkedBlockingDeque 是一个基于链表结构的双端阻塞队列。它同时支持从队列头部插入和移除元素，也支持从队列尾部插入和移除元素。因此，LinkedBlockingDeque 可以作为 FIFO 队列或 LIFO 队列来使用。

常用方法有：

addFirst(E e), addLast(E e): 在队列的开头/结尾添加元素。
takeFirst(), takeLast(): 从队列的开头/结尾移除和返回元素，如果队列为空，则等待。
putFirst(E e), putLast(E e): 在队列的开头/结尾插入元素，如果队列已满，则等待。
pollFirst(long timeout, TimeUnit unit), pollLast(long timeout, TimeUnit unit): 在队列的开头/结尾移除和返回元素，如果队列为空，则等待指定的超时时间。

使用示例：

public class LinkedBlockingDequeDemo { public static void main(String[] args) throws InterruptedException { LinkedBlockingDeque<String> deque = new LinkedBlockingDeque<>(10); // Adding elements at the end of the deque deque.putLast("Item1"); deque.putLast("Item2"); // Adding elements at the beginning of the deque deque.putFirst("Item3"); // Removing elements from the beginning System.out.println(deque.takeFirst()); // Output: Item3 // Removing elements from the end System.out.println(deque.takeLast()); // Output: Item2 } }

程序运行结果如下：

DelayQueue

DelayQueue 是一个无界阻塞队列，用于存放实现了 Delayed 接口的元素，这些元素只能在其到期时才能从队列中取走。这使得 DelayQueue 成为实现时间基于优先级的调度服务的理想选择。

下面的代码案例展示了如何使用 DelayQueue。

public class DelayQueueDemo { public static void main(String[] args) { DelayQueue<DelayedElement> queue = new DelayQueue<>(); // 将带有5秒延迟的元素放入队列 queue.put(new DelayedElement(8000, "这是一个 8 秒延迟的元素")); try { System.out.println("取一个元素..."); // take() 将阻塞，直到延迟到期 DelayedElement element = queue.take(); System.out.println(element.getMessage()); } catch (InterruptedException e) { e.printStackTrace(); } } static class DelayedElement implements Delayed { private final long delayUntil; private final String message; public DelayedElement(long delayInMillis, String message) { this.delayUntil = System.currentTimeMillis() + delayInMillis; this.message = message; } public String getMessage() { return message; } @Override public long getDelay(TimeUnit unit) { return unit.convert(delayUntil - System.currentTimeMillis(), TimeUnit.MILLISECONDS); } @Override public int compareTo(Delayed o) { return Long.compare(this.delayUntil, ((DelayedElement) o).delayUntil); } } }

上例创建了一个 DelayQueue，并将一个带有 5 秒延迟的元素放入队列。然后，它调用 take() 方法从队列中取出元素。由于元素的延迟时间为 5 秒，因此 take() 方法将阻塞 5 秒，直到元素到期。运行结果如下：