阻塞队列 BlockingQueue

文章目录

  • 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 通知机制来实现可阻塞的插入和删除。

不同点

  1. ArrayBlockingQueue 基于数组实现,而 LinkedBlockingQueue 基于链表实现;
  2. 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();}
}