文章目录
- BlockingQueue 阻塞队列
- 一、BlockingQueue 接口
- 1、概述
- 2、接口方法
- 3、小结
- 二、ArrayBlockingQueue
- 1、概述
- 2、基本实现
- 3、put 方法详解
- 4、take 方法详解
- 5、使用示例
- 三、LinkedBlockingQueue
- 1、概述
- 2、基本实现
- 3、put 方法详解
- 4、take 方法详解
- 5、对比 ArrayBlockingQueue
- 6、使用示例
- 四、PriorityBlockingQueue
- 五、SynchronousQueue
- 六、DelayQueue
- 七、LinkedBlockingDeque
- 八、LinkedTransferQueue
BlockingQueue 阻塞队列
一、BlockingQueue 接口
1、概述
BlockingQueue 代表了一个线程安全的阻塞队列,不仅可以由多个线程并发访问,还提供了可阻塞的插入和移除方法。
- 当队列已满,向队列中
put添加元素时(生产)会被阻塞,直到队列有足够的空间。 - 当队列为空,从队列中
take移除元素时(消费)会被阻塞,直到队列有元素可用。
因此,BlockingQueue 被广泛用于“生产者-消费者”问题中。
2、接口方法
BlockingQueue 接口继承于 Queue 接口
public interface Queue<E> extends Collection<E> {// 将指定的元素插入到队列中,如果队列已满,则抛出 IllegalStateException 异常。boolean add(E e);// 将指定的元素插入到队列中,如果队列已满,则返回 false,否则返回 true。boolean offer(E e);// 移除并返回队列的头部元素,如果队列为空,则抛出 NoSuchElementException 异常。E remove();// 移除并返回队列的头部元素,如果队列为空,则返回 null。E poll();// 获取但不移除队列的头部元素,如果队列为空,则抛出 NoSuchElementException 异常。E element();// 获取但不移除队列的头部元素,如果队列为空,则返回 null。E peek();
}
BlockingQueue 接口还扩展了以下方法(可以看到有阻塞的插入和移除方法)
public interface BlockingQueue<E> extends Queue<E> {// 将指定的元素插入到队列中,如果队列已满,则阻塞等待,直到队列有足够的空间。void put(E e) throws InterruptedException;// 移除并返回队列的头部元素,如果队列为空,则阻塞等待,直到队列有元素可用。E take() throws InterruptedException;// 带超时时间的offerboolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;// 带超时时间的pollE poll(long timeout, TimeUnit unit) throws InterruptedException;// 从阻塞队列中移除并返回指定数量的元素,并将它们添加到给定的集合中。int drainTo(Collection<? super E> c);// maxElements:要移除的最大元素数量。int drainTo(Collection<? super E> c, int maxElements);// 返回当前阻塞队列中剩余的可用容量,即队列中可以插入的元素数量。int remainingCapacity();
}
3、小结
总结一下 BlockingQueue 主要的几个方法:
| 方式 | 抛出异常 | 不抛异常,返回false或null | 阻塞等待 | 超时等待 |
|---|---|---|---|---|
| 添加 | add(E e) | offer(E e) | put(E e) | offer(e, timeout, unit) |
| 移除并返回首个 | remove() | poll() | take() | poll(timeout, unit) |
| 获取不移除首个 | element() | peek() | — | — |
二、ArrayBlockingQueue
1、概述
ArrayBlockingQueue 是一个基于数组的有界阻塞队列:
- 有界:在构造时就确定了容量大小,并且在之后不能更改。这个界限提供了流量控制,有助于资源的合理使用。
- FIFO:队列操作符合先进先出的原则。
- 公平:构造方法提供的
boolean fair参数,可以指定是否公平。
2、基本实现
// java.util.concurrent.ArrayBlockingQueue/** The queued items */
final Object[] items;/** items index for next take, poll, peek or remove */
int takeIndex;/** items index for next put, offer, or add */
int putIndex;/** Number of elements in the queue */
int count;/** Concurrency control uses the classic two-condition algorithm* found in any textbook.*//** Main lock guarding all access */
final ReentrantLock lock;/** Condition for waiting takes */
private final Condition notEmpty;/** Condition for waiting puts */
private final Condition notFull;public ArrayBlockingQueue(int capacity, boolean fair) {if (capacity <= 0) throw new IllegalArgumentException();this.items = new Object[capacity];lock = new ReentrantLock(fair);notEmpty = lock.newCondition();notFull = lock.newCondition();
}
3、put 方法详解
// java.util.concurrent.ArrayBlockingQueuepublic void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {// 如果当前队列已满,将线程移入到notFull等待队列中while (count == items.length)notFull.await();// 满足插入数据的要求,直接进行入队操作enqueue(e);} finally {lock.unlock();}
}private void enqueue(E x) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;final Object[] items = this.items;// 插入数据items[putIndex] = x;if (++putIndex == items.length)putIndex = 0;count++;// 唤醒阻塞的消费者线程,当前队列中有数据可供消费notEmpty.signal();
}
4、take 方法详解
// java.util.concurrent.ArrayBlockingQueuepublic E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {// 如果当前队列为空,将线程移入到notEmpty等待队列中while (count == 0)notEmpty.await();// 获取数据return dequeue();} finally {lock.unlock();}
}private E dequeue() {final Object[] items = this.items;@SuppressWarnings("unchecked")// 获取数据E x = (E) items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();// 唤醒阻塞的生产者线程,队列已经空出来了。notFull.signal();return x;
}
5、使用示例
public class ArrayBlockingQueueDemo {public static void main(String[] args) {BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);// Producernew Thread(() -> {try {for (int i = 0; i < 10; i++) {queue.put(i);System.out.println("Produced: " + i);Thread.sleep(1000);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();// Consumernew Thread(() -> {try {for (int i = 0; i < 10; i++) {int value = queue.take();System.out.println("Consumed: " + value);Thread.sleep(2000);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();}
}
三、LinkedBlockingQueue
1、概述
LinkedBlockingQueue 是一个基于链表的有界/无界阻塞队列:
- 可以在队列头部和尾部进行高效的插入和删除操作。
- 可以在构造时指定最大容量。如果不指定,默认为
Integer.MAX_VALUE,这意味着队列的大小受限于可用内存。
2、基本实现
// java.util.concurrent.LinkedBlockingQueue/** Current number of elements */
private final AtomicInteger count = new AtomicInteger();/*** Head of linked list.* Invariant: head.item == null*/
transient Node<E> head;/*** Tail of linked list.* Invariant: last.next == null*/
private transient Node<E> last;/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();static class Node<E> {// 头节点的 item 始终为 null,它作为一个虚拟节点,用于帮助管理队列。E item;// 尾节点的 next 始终为 null,因为没有下一个元素了Node<E> next;Node(E x) { item = x; }
}
3、put 方法详解
// java.util.concurrent.LinkedBlockingQueuepublic void put(E e) throws InterruptedException {if (e == null) throw new NullPointerException();// 用于存储操作前的队列元素数量,预设为 -1 表示失败,除非稍后设置。int c = -1;// 创建一个新的节点包含要插入的元素 e。Node<E> node = new Node<E>(e);// 获取put锁和计数器。final ReentrantLock putLock = this.putLock;final AtomicInteger count = this.count;putLock.lockInterruptibly();try {// 如果队列已满,则阻塞当前线程,将其移入等待队列while (count.get() == capacity) {notFull.await();}// 入队操作,插入数据enqueue(node);c = count.getAndIncrement();// 队列没有满,则通知被阻塞的生产者线程if (c + 1 < capacity)notFull.signal();} finally {putLock.unlock();}// 如果插入操作将队列从空变为非空,唤醒可能正在等待非空队列的消费者线程。if (c == 0)signalNotEmpty();
}private void enqueue(Node<E> node) {// assert putLock.isHeldByCurrentThread();// assert last.next == null;last = last.next = node;
}private void signalNotEmpty() {final ReentrantLock takeLock = this.takeLock;takeLock.lock();try {notEmpty.signal();} finally {takeLock.unlock();}
}
4、take 方法详解
// java.util.concurrent.LinkedBlockingQueuepublic E take() throws InterruptedException {// 用于存储被取出的元素。E x;// 用于存储操作前的队列元素数量,预设为 -1 表示失败,除非稍后设置。int c = -1;// 获取计数器和take锁final AtomicInteger count = this.count;final ReentrantLock takeLock = this.takeLock;takeLock.lockInterruptibly();try {// 当前队列为空,则阻塞当前线程,将其移入到等待队列中,直至满足条件while (count.get() == 0) {notEmpty.await();}// 移除队头元素,获取数据x = dequeue();c = count.getAndDecrement();// 如果当前满足移除元素的条件,则通知被阻塞的消费者线程if (c > 1)notEmpty.signal();} finally {takeLock.unlock();}if (c == capacity)signalNotFull();return x;
}private E dequeue() {Node<E> h = head;Node<E> first = h.next;h.next = h; // help GChead = first;E x = first.item;first.item = null;return x;
}private void signalNotFull() {final ReentrantLock putLock = this.putLock;putLock.lock();try {notFull.signal();} finally {putLock.unlock();}
}
5、对比 ArrayBlockingQueue
相同点:
ArrayBlockingQueue 和 LinkedBlockingQueue 都是通过 Condition 通知机制来实现可阻塞的插入和删除。
不同点:
- ArrayBlockingQueue 基于数组实现,而 LinkedBlockingQueue 基于链表实现;
- ArrayBlockingQueue 使用一个单独的 ReentrantLock 来控制对队列的访问,而 LinkedBlockingQueue 使用两个锁(putLock 和 takeLock),一个用于放入操作,另一个用于取出操作。这可以提供更细粒度的控制,并可能减少线程之间的竞争。
6、使用示例
public class LinkedBlockingQueueDemo {public static void main(String[] args) {BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(5);// Producernew Thread(() -> {try {for (int i = 0; i < 10; i++) {queue.put(i);System.out.println("Produced: " + i);Thread.sleep(1000);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();// Consumernew Thread(() -> {try {for (int i = 0; i < 10; i++) {int value = queue.take();System.out.println("Consumed: " + value);Thread.sleep(2000);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();}
}
四、PriorityBlockingQueue
PriorityBlockingQueue 是一个具有优先级排序特性的无界阻塞队列。
- 可以在构造方法中指定
Comparator比较器进行定制排序。 - 如果没有指定,遵循自然排序(如果队列存储的元素已经实现了
Comparable,会使用实现的)
当需要根据优先级来执行任务时,PriorityBlockingQueue 会非常有用。
【构造函数】
// java.util.concurrent.PriorityBlockingQueuepublic PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator) {if (initialCapacity < 1) throw new IllegalArgumentException();this.lock = new ReentrantLock();this.notEmpty = lock.newCondition();this.comparator = comparator;this.queue = new Object[initialCapacity];
}
【使用示例】
class Task implements Comparable<Task> {private int priority;private String name;public Task(int priority, String name) {this.priority = priority;this.name = name;}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"));TimeUnit.MILLISECONDS.sleep(100);queue.put(new Task(10, "Medium priority task"));TimeUnit.MILLISECONDS.sleep(100);queue.put(new Task(50, "High priority task"));while (!queue.isEmpty()) {System.out.println(queue.take().getName());}}
}
五、SynchronousQueue
SynchronousQueue 是一个非常特殊的阻塞队列,它不存储任何元素。
- 每一个插入操作必须等待另一个线程的移除操作,反之亦然。
因此,SynchronousQueue 的内部实际上是空的,但它允许一个线程向另一个线程逐个传输元素。
【使用示例】
public class SynchronousQueueDemo {public static void main(String[] args) {SynchronousQueue<String> queue = new SynchronousQueue<>();new Thread(() -> {try {System.out.println(Thread.currentThread().getName() + " put a");queue.put("a");System.out.println(Thread.currentThread().getName() + " put b");queue.put("b");} catch (InterruptedException e) {e.printStackTrace();}}, "A").start();new Thread(() -> {try {TimeUnit.SECONDS.sleep(2);String a = queue.take();System.out.println(Thread.currentThread().getName() + " take " + a);TimeUnit.SECONDS.sleep(2);String b = queue.take();System.out.println(Thread.currentThread().getName() + " take " + b);} catch (InterruptedException e) {// Todo Handle Exception}}, "B").start();}
}
put 了一个元素,必须先 take 取出来,否则不能再 put 进去值(存一个取一个,最多一个)
六、DelayQueue
DelayQueue 是一个无界阻塞队列,用于存放实现了 Delayed 接口的元素,这些元素只能在其到期时才能从队列中取走。这使得 DelayQueue 成为实现时间基于优先级的调度服务的理想选择。
【使用示例】
public class DelayQueueDemo {public static void main(String[] args) {DelayQueue<DelayedElement> queue = new DelayQueue<>();// 将带有5秒延迟的元素放入队列queue.put(new DelayedElement(5000, "这是一个 5 秒延迟的元素"));try {System.out.println("取一个元素...");// take() 将阻塞,直到延迟到期DelayedElement element = queue.take();System.out.println(element.getMessage());} catch (InterruptedException e) {// Todo Handle Exception}}/*** DelayQueueDemo存储的元素,需要实现Delayed接口*/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;}@Overridepublic long getDelay(TimeUnit unit) {return unit.convert(delayUntil - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}@Overridepublic int compareTo(Delayed o) {return Long.compare(this.delayUntil, ((DelayedElement) o).delayUntil);}}
}
七、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 dequedeque.putLast("Item1");deque.putLast("Item2");// Adding elements at the beginning of the dequedeque.putFirst("Item3");// Removing elements from the beginningSystem.out.println(deque.takeFirst()); // Output: Item3// Removing elements from the endSystem.out.println(deque.takeLast()); // Output: Item2}
}
运行结果如下:
Item3
Item2
八、LinkedTransferQueue
LinkedTransferQueue 是一个基于链表结构的无界传输队列,
LinkedTransferQueue 实现了 TransferQueue 接口,提供了一种强大的线程间交流机制
- 允许一个元素直接从生产者传输给消费者,如果消费者已经在等待。如果没有等待的消费者,元素将入队。
public interface TransferQueue<E> extends BlockingQueue<E> {// 尝试立即转移元素,如果有消费者正在等待,则传输成功;否则,返回 false。boolean tryTransfer(E e);// 将元素转移到等待的消费者,如果不存在等待的消费者,则元素会入队并阻塞直到该元素被消费。void transfer(E e) throws InterruptedException;boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;
}
如果想要更紧密地控制生产者和消费者之间的交互,可以使用 LinkedTransferQueue。
【使用示例】
public class LinkedTransferQueueDemo {public static void main(String[] args) {LinkedTransferQueue<Integer> queue = new LinkedTransferQueue<>();// Producernew Thread(() -> {try {for (int i = 0; i < 10; i++) {queue.transfer(i); // 将元素传输给消费者线程System.out.println("Produced: " + i);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();// Consumernew Thread(() -> {try {for (int i = 0; i < 10; i++) {TimeUnit.SECONDS.sleep(1);int value = queue.take();System.out.println("Consumed: " + value);}} catch (InterruptedException e) {// Todo Handle Exception}}).start();}
}