Java 多线程并发【12】BlockingQueue 体系

本文源码分析基于 https://github.com/openjdk/jdk

BlockingQueue

BlockingQueue ，阻塞队列，通常用于生产者消费者模式。

阻塞队列的原理

BlockingQueue 接口继承自 Queue ，通过泛型指定队列中的元素类型。当生产者产生元素时，会将元素插入队列尾部；当消费者需要消费元素时，从队列头部取出元素。

队列中的元素有最大容纳的数量，当达到这个临界点时，生产者插入新对象时会阻塞，直到有消费者从队列中取出元素；当队列为空时，消费者无法从空队列中提取元素，那么消费者就会进入阻塞状态，直到队列中有新的元素出现。

代码分析

public interface BlockingQueue<E> extends Queue<E> {

    boolean add(E e);

    boolean offer(E e);

    void put(E e) throws InterruptedException;

    boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException;

    E take() throws InterruptedException;

    E poll(long timeout, TimeUnit unit)
            throws InterruptedException;

    int remainingCapacity();

    boolean remove(Object o);

    boolean contains(Object o);

    int drainTo(Collection<? super E> c);

    int drainTo(Collection<? super E> c, int maxElements);
}

BlockingQueue 继承自 Queue 接口，Queue 接口中的方法包括：

public interface Queue<E> extends Collection<E> {
  // 添加到队列头部
    boolean add(E e);
    // 添加到队列头部，插入元素不可为空，为空抛出 NPE
    boolean offer(E e);
    // 检索并删除该队列头部。此方法与poll的不同之处在于，如果队列为空，则会抛出异常。
    E remove();
  // 检索并删除该队列的头部，如果该队列为空，则返回null。
    E poll();
  // 检索但不删除该队列的头部。这个方法与peek的不同之处：如果队列为空，它会抛出异常。
    E element();
  // 检索但不删除该队列的头部。这个方法与peek的不同之处：如果队列为空，则返回null。
    E peek();
}

BlockingQueue 中拓展了几个新方法，包括：

    void put(E e) throws InterruptedException;
    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
    E take() throws InterruptedException;
    E poll(long timeout, TimeUnit unit) throws InterruptedException;
    int remainingCapacity();
    boolean remove(Object o);
    boolean contains(Object o);
    int drainTo(Collection<? super E> c);
    int drainTo(Collection<? super E> c, int maxElements);

BlockingQueue 的方法有四组：

	抛出异常	特殊值	阻塞	超时
Insert	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
Remove	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
Examine	element(o)	peek(o)

抛出异常：如果尝试操作无法立即执行，直接抛出异常。
特殊值：如果尝试的操作无法立即执行，返回特定的值，如 null 或 false ，取决于操作。
阻塞：如果尝试的操作无法立即执行，则进入阻塞状态，直到能够执行。
超时：如果尝试的操作无法立即执行，则进入阻塞状态，直到能够执行，但有最大等待时间限制，超时返回一个特定的值告知是否成功（一般是 true/false）。

其他特性

这部分来自于官方文档：

BlockingQueue 不接受空元素，add/put/offer 方法接收一个 null 时，会抛出 NullPointerException 。null 用作标志值，表示轮询操作失败。
BlockingQueue 可能有容量限制。在任何给定的时间，它可能有一个剩余容量，超过这个容量就不能放置任何额外的元素而不会阻塞。没有任何内在容量限制的 BlockingQueue 总是报告 Integer.MAX_VALUE 的剩余容量。
BlockingQueue 实现主要用于生产者-消费者队列，但还支持 Collection 接口。例如，可以使用 remove(x) 从队列中删除任意元素。然而，这样的操作通常执行起来效率不高，并且只用于偶尔的使用，比如取消排队的消息。
BlockingQueue 实现是线程安全的。所有排队方法都使用内部锁或其他形式的并发控制原子地实现它们的效果。但是，批量收集操作 addAll, containsAll, retainAll 和 removeAll 不一定是原子执行的，除非在实现中另外指定。因此，例如，addAll(c) 在只添加了 c 中的一些元素之后就可能失败(抛出异常)。

BlockingDeque

BlockingDeque 用来定义双端队列，不同的线程可以从这个双端队列中提取元素。它的阻塞原理和 BlockingQueue 是一样的。

在不能插入元素时，将阻塞试图插入元素的线程。
在不能获取元素时，将阻塞试图获取元素的线程。

代码分析

public interface BlockingDeque<E> extends BlockingQueue<E>, Deque<E> {

    void addFirst(E e);

    void addLast(E e);

    boolean offerFirst(E e);

    boolean offerLast(E e);

    void putFirst(E e) throws InterruptedException;

    void putLast(E e) throws InterruptedException;
    
    boolean offerFirst(E e, long timeout, TimeUnit unit) throws InterruptedException;
    
    boolean offerLast(E e, long timeout, TimeUnit unit) throws InterruptedException;
    
    E takeFirst() throws InterruptedException;

    E takeLast() throws InterruptedException;

    E pollFirst(long timeout, TimeUnit unit)
            throws InterruptedException;

    E pollLast(long timeout, TimeUnit unit)
            throws InterruptedException;

    boolean removeFirstOccurrence(Object o);

    boolean removeLastOccurrence(Object o);

    // *** BlockingQueue methods ***
    boolean add(E e);

    boolean offer(E e);
    
    void put(E e) throws InterruptedException;

    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;

    E remove();

    E poll();

    E take() throws InterruptedException;

    E poll(long timeout, TimeUnit unit) throws InterruptedException;

    E element();

    E peek();

    boolean remove(Object o);
    
    public boolean contains(Object o);
    
    public int size();

    Iterator<E> iterator();

    // *** Stack methods ***
    void push(E e);
}

BlockingDeque 中的方法一部分来自于 Deque 接口，可以按照对头节点和尾节点的操作分为两组：

First Element (Head)

	Throws exception	Special value	Blocks	Times out
Insert	addFirst(e)	offerFirst(e)	putFirst(e)	offerFirst(e, time, unit)
Remove	removeFirst()	pollFirst()	takeFirst()	pollFirst(time, unit)
Examine	getFirst()	peekFirst()

Last Element (Tail)

	Throws exception	Special value	Blocks	Times out
Insert	addLast(e)	offerLast(e)	putLast(e)	offerLast(e, time, unit)
Remove	removeLast()	pollLast()	takeLast()	pollLast(time, unit)
Examine	getLast()	peekLast()

BlockingDeque 实现可以直接作为 FIFO BlockingQueue 使用。从 BlockingQueue 接口继承的方法完全等价于 BlockingDeque 方法，如下表所示:

BlockingDeque 和 BlockingQueue 相等的方法

BlockingQueue	BlockingDeque
add(e)	addLast(e)
offer(e)	offerLast(e)
put(e)	putLast(e)
offer(e, time, unit)	offerLast(e, time, unit)
remove()	removeFirst()
poll()	pollFirst()
take()	takeFirst()
poll(time, unit)	pollFirst(time, unit)
element()	getFirst()
peek()	peekFirst()

BlockingQueue 的实现

ArrayBlockingQueue

ArrayBlockingQueue 数组阻塞队列，是一个有界的阻塞队列，内部数据结构是一个数组，队列有存储数量的上限，它的构造方法中可以直接指定最大容纳数量：

final Object[] items; // 内部的数据结构

public ArrayBlockingQueue(int capacity) {
   this(capacity, false);
}

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();
}

内部实现了线程安全保障，通过 ReentrantLock 实现的同步。

DelayQueue

DelayQueue ，延迟队列，持有元素一段时间，直到到达时间后外部才能获取这个元素。

DelayQueue 继承自 AbstractQueue ，并且它的元素是必须实现 Delayed 接口：

public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E>
  
  
public interface Delayed extends Comparable<Delayed> {
    /**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     */
    long getDelay(TimeUnit unit);
}

线程安全通过 ReentrantLock 实现。

LinkedBlockingQueue

LinkedBlockingQueue 链表阻塞队列，内部的数据结构是链表，并且可以设置一个容量上限，如果不设置，默认是 Integer.MAX_VALUE 。

transient Node<E> head; // 数据结构

private transient Node<E> last;

public LinkedBlockingQueue() {
  this(Integer.MAX_VALUE);
}

public LinkedBlockingQueue(int capacity) {
  if (capacity <= 0) throw new IllegalArgumentException();
  this.capacity = capacity;
  last = head = new Node<E>(null);
}

节点类型是：

static class Node<E> {
  E item;
  Node<E> next;
  Node(E x) { item = x; }
}

这个类的线程安全也是通过 ReentrantLock 保障的：

private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();

PriorityBlockingQueue

PriorityBlockingQueue 优先级阻塞队列，一个无界阻塞队列，它使用与类 PriorityQueue 相同的排序规则，并提供阻塞检索操作。虽然该队列在逻辑上是无界的，但尝试添加可能会由于资源耗尽而失败(导致 OutOfMemoryError )。这个类不允许有空元素。依赖于自然排序的优先级队列也不允许插入不可比较的对象(这样做会导致 ClassCastException )。

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable

内部数据结构是数组：

private transient Object[] queue;

构造方法可以设置初始容量，也可以设置排序规则：

public PriorityBlockingQueue() {
  this(DEFAULT_INITIAL_CAPACITY, null); // DEFAULT_INITIAL_CAPACITY = 11
}

public PriorityBlockingQueue(int initialCapacity) {
  this(initialCapacity, null);
}

public PriorityBlockingQueue(int initialCapacity,
                                 Comparator<? super E> comparator) {
  if (initialCapacity < 1) throw new IllegalArgumentException();
  this.comparator = comparator;
  this.queue = new Object[Math.max(1, initialCapacity)];
}

线程安全通过 ReentrantLock 实现。、

SynchronousQueue

SynchronousQueue 同步队列，只能容纳一个元素，每个插入操作都必须等待另一个线程的取出操作。

这个类支持一个可选的公平性策略，用于对等待的生产者和消费者线程进行排序。默认情况下，不保证此顺序。然而，将公平性设置为true的队列以FIFO顺序授予线程访问权。这个类及其迭代器实现了Collection和iterator接口的所有可选方法。

数据结构通过 Transferer 接口实例来管理，它有两个实现类：

    abstract static class Transferer<E> {
        abstract E transfer(E e, boolean timed, long nanos);
    }
    
    static final class TransferStack<E> extends Transferer<E> 
    
    static final class TransferQueue<E> extends Transferer<E>

具体细节这里不详细介绍了，其复杂度可以单独讲了。这个的线程安全不是通过 ReentrantLock 实现的，而是自旋形式的阻塞。

BlockingDeque 的实现

LinkedBlockingDeque

LinkedBlockingDeque 链表双端阻塞队列，相当于 LinkedBlockingQueue 的 Deque 版本。内部节点结构：

    static final class Node<E> {
        E item;

        Node<E> prev;
        
        Node<E> next;

        Node(E x) {
            item = x;
        }
    }

线程安全保障：

    /** Main lock guarding all access */
    final ReentrantLock lock = new ReentrantLock();

    /** Condition for waiting takes */
    @SuppressWarnings("serial") // Classes implementing Condition may be serializable.
    private final Condition notEmpty = lock.newCondition();

    /** Condition for waiting puts */
    @SuppressWarnings("serial") // Classes implementing Condition may be serializable.
    private final Condition notFull = lock.newCondition();