Java工程师知识树 / Java基础

队列介绍

多线程应用中，队列的使用率很高，多数生产消费模型的首选数据结构就是队列(FIFO先进先出)。按照实现方式的不同可分为阻塞队列和非阻塞队列，前者使用锁实现，而后者则使用 CAS 非阻塞算法实现。

在concurrent包中，提供了多种BlockingQueue的实现，对比常用的几个：

LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue实现了BlockingQueue的接口，也间接的实现了Queue接口，它们都是阻塞队列。

ConcurrentLinkedQueue没有实现BlockingQueue接口，而是实现了Queue接口，它是一个非阻塞模式的队列。

BlockingQueue接口

BlockingQueue接口提供了几个主要的方法，主要分为添加元素与删除元素两大类：

LinkedBlockingQueue

首先，我们来看一下LinkedBlockingQueue。LinkedBlockingQueue是一个基于链表结构，大小无界的，可以不指定队列的大小的阻塞队列。它按照FIFO（先进先出）的方式排序元素。

说是大小无界的队列，但其实它的默认大小是有指定的，为Integer的最大值，因此也可以认为是大小无界的。当然我们也可以自行指定其初始化大小。

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

LinkedBlockingQueue内部维护了一个单链表数据结构，用于存储元素：

static class Node<E> {
    E item;

    Node<E> next;

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

transient Node<E> head;

private transient Node<E> last;

同时，为了线程安全的考虑，它在内部维护了两把锁，分别用于读取元素与写入元素：

private final ReentrantLock takeLock = new ReentrantLock();

private final Condition notEmpty = takeLock.newCondition();

private final ReentrantLock putLock = new ReentrantLock();

private final Condition notFull = putLock.newCondition();

使用读锁与写锁进行分类，可以让生产操作与消费操作并行运行。

我们抽出两个比较常用的方法，offer()与poll()，对其进行分析。

offer()

首先来看一下offer()的实现：

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}

private void enqueue(Node<E> node) {
    last = last.next = node;
}

上面就是LinkedBlockingQueue中offer()方法的实现，整体的逻辑比较简单，主要分为几步：

1、检查队列元素数量是否已经达到阈值，如果达到，返回false

2、获取写锁，成功后，将元素加入队列尾部，元素数加一

3、检查当前元素数加一后的数值，是否小于阈值，如果是，唤醒写锁上等待的线程

4、释放写锁

5、如果该元素是队列中第一个元素，唤醒读锁上等待的线程（如果有）

offer()方法同时也支持带过期时间的参数，在过期时间内，如果队列元素仍满，则返回false。

poll()

接下来，我们再看一下获取元素的方法，poll()的实现：

public E poll() {
    final AtomicInteger count = this.count;
    if (count.get() == 0)
        return null;
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        if (count.get() > 0) {
            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 GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

poll()的实现也比较清晰，简单说明一下：

1、检查当前队列中是否有元素，如果没有，返回null

2、获取读锁，成功后，检查队列中元素个数是否大于0，如果是，将头部元素移除队列

3、检查移除后，元素个数是否大于0，如果是，唤醒读锁上的等待线程（如果有）

4、释放读锁

5、检查移除前的元素个数是否达到阈值，如果是，那么说明线程队列中有空间，唤醒写锁上的等待线程

6、返回移除的元素

poll()同样也提供了带超时时间的参数方法，实现基本与offer()一致，这里不再赘述。

ArrayBlockingQueue

上面我们看完了LinkedBlockingQueue的实现，现在我们再来看一下ArrayBlockingQueue，ArrayBlockingQueue是一个有界的阻塞队列，初始化时必须指定其容量大小：

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

它与LinkedBlockingQueue的内部存储实现并不相同，其内部采用了数组结构存储元素，同时它在内部只维护了一把锁，即读元素与写元素，共用一把锁，在并发场景下的吞吐量，会低于LinkedBlockingQueue。

final Object[] items;

int takeIndex;

int putIndex;

int count;

final ReentrantLock lock;

private final Condition notEmpty;

private final Condition notFull;

在构造函数中，我们可以指定ArrayBlockingQueue的访问策略，即指定使用公平锁或非公平锁。

默认情况下，使用的是非公平锁的模式，这样可以获得更好的吞吐量，但不能严格保证生产者线程与消费者线程的顺序；如果使用公平锁的模式，那么队列允许按照 FIFO 顺序访问线程，但牺牲的将会是吞吐量。

同样的，我们也对两个比较常用的方法，offer()与poll()的实现进行简要分析。

offer()

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
        return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

poll()

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : 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;
}

上面就是ArrayBlockingQueue的offer()与poll()的实现，大体上与LinkedBlockingQueue差不多。

PriorityBlockingQueue

PriorityBlockingQueue是一个基于元素优先级的无界阻塞队列。虽然此队列逻辑上是无界的，但是资源被耗尽时试图执行 add 操作也将失败。

与ArrayBlockingQueue一样，其内部采用数组结构存储元素，有一点比较特殊的是，PriorityBlockingQueue是不允许插入不可比较的对象的，会抛出ClassCastException，因此，在队列中的元素，都需要实现Comparator接口。

我们来看一下其内部结构：

// 默认大小
private static final int DEFAULT_INITIAL_CAPACITY = 11;

// 最大容量
private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

private transient Object[] queue;

private transient int size;

private transient Comparator<? super E> comparator;

private final ReentrantLock lock;

private final Condition notEmpty;

private transient volatile int allocationSpinLock;

其内部与ArrayBlockingQueue一样，都是只使用了一把锁，用于写入元素与读取元素。

PriorityBlockingQueue的默认初始化大小为11，当达到阈值值，会进行扩容操作，扩大50%。

我们也可以在创建时，自行指定初始化大小，如果你可以预先估算出所需容量，那么初始化时直接进行指定是比较推荐的。在初始化时，也可以传入一个 Comparator比较器对象，队列中的元素会根据比较器对元素进行排序。

接下来还是看一下最常用的offer()与poll()方法。

offer()

public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lock();
    int n, cap;
    Object[] array;
    while ((n = size) >= (cap = (array = queue).length))
        tryGrow(array, cap);
    try {
        Comparator<? super E> cmp = comparator;
        if (cmp == null)
            siftUpComparable(n, e, array);
        else
            siftUpUsingComparator(n, e, array, cmp);
        size = n + 1;
        notEmpty.signal();
    } finally {
        lock.unlock();
    }
    return true;
}

private static <T> void siftUpComparable(int k, T x, Object[] array) {
    Comparable<? super T> key = (Comparable<? super T>) x;
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        Object e = array[parent];
        if (key.compareTo((T) e) >= 0)
            break;
        array[k] = e;
        k = parent;
    }
    array[k] = key;
}

private static <T> void siftUpUsingComparator(int k, T x, Object[] array, Comparator<? super T> cmp) {
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        Object e = array[parent];
        if (cmp.compare(x, (T) e) >= 0)
            break;
        array[k] = e;
        k = parent;
    }
    array[k] = x;
}

我们简要说明一下offer()方法所进行的操作：

• 1、判断元素是否非空
• 2、获取锁
• 3、判断当前队列容量是否达到阈值，如果是，进行扩容操作
• 4、将元素根据堆排序的结果，放入数组指定位置
• 5、数组容量计数器+1
• 6、释放锁

这里比较有趣的是关于元素存放的算法，这里采用的是最小堆的算法，将数组中的元素构建成了一棵二叉树，每次放入确保比较结果最小的那个元素，放在堆顶，每次加入新元素时，会重新构建堆。

再看一下poll()。

poll()

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return dequeue();
    } finally {
        lock.unlock();
    }
}

private E dequeue() {
    int n = size - 1;
    if (n < 0)
        return null;
    else {
        Object[] array = queue;
        E result = (E) array[0];
        E x = (E) array[n];
        array[n] = null;
        Comparator<? super E> cmp = comparator;
        if (cmp == null)
            siftDownComparable(0, x, array, n);
        else
            siftDownUsingComparator(0, x, array, n, cmp);
        size = n;
        return result;
    }
}


private static <T> void siftDownComparable(int k, T x, Object[] array, int n) {
    if (n > 0) {
        Comparable<? super T> key = (Comparable<? super T>) x;
        int half = n >>> 1;           // loop while a non-leaf
        while (k < half) {
            int child = (k << 1) + 1; // assume left child is least
            Object c = array[child];
            int right = child + 1;
            if (right < n &&
                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
                c = array[child = right];
            if (key.compareTo((T) c) <= 0)
                break;
            array[k] = c;
            k = child;
        }
        array[k] = key;
    }
}

poll()方法的逻辑如上，其实做的事情比较简单，将堆顶的元素拿出，然后重新将剩余元素构建最小堆。

ConcurrentLinkedQueue

ConcurrentLinkedQueue是一个基于链表结构存储的，无界的，线程安全的队列，它按照 FIFO（先进先出）原则对元素进行排序。当新增一个元素时，会插入到队列的尾部，获取一个元素时，会返回头部的元素。

ConcurrentLinkedQueue是一个线程安全的队列，但是其内部并没有像LinkedBlockingQueue那样使用锁来实现，而是基于CAS操作的”无等待 (wait-free)”算法来实现的线程安全。

ConcurrentLinkedQueue 的非阻塞算法实现主要可概括为下面几点：

• 1、使用 CAS 原子指令来处理对数据的并发访问，这是非阻塞算法得以实现的基础。
• 2、head/tail 并非总是指向队列的头 / 尾节点，也就是说允许队列处于不一致状态。 这个特性把入队 / 出队时，原本需要一起原子化执行的两个步骤分离开来，从而缩小了入队 / 出队时需要原子化更新值的范围到唯一变量。这是非阻塞算法得以实现的关键。
• 3、以批处理方式来更新head/tail，从整体上减少入队 / 出队操作的开销。

ConcurrentLinkedQueue 内部采用链表结构存储元素，基本结构与LinkedBlockingQueue一致，不同的是，在设置链表元素的内容时，采用的全部是CAS的操作，保证了操作的原子性。

注意的是，ConcurrentLinkedQueue 的存储的元素不可以为null。

多种队列对比

不同的队列使用与不同的场景，几种队列的特点与适应场景汇总与对比：