并发编程4：Java 阻塞队列源码分析（上）
并发编程4：Java 阻塞队列源码分析（上）
上篇文章 并发编程3：线程池的使用与执行流程 中我们了解到，线程池中需要使用阻塞队列来保存待执行的任务。这篇文章我们来详细了解下 Java 中的阻塞队列究竟是什么。
什么是阻塞队列
阻塞队列其实就是生产者-消费者模型中的容器。
当生产者往队列中添加元素时，如果队列已经满了，生产者所在的线程就会阻塞，直到消费者取元素时 notify 它； 消费者去队列中取元素时，如果队列中是空的，消费者所在的线程就会阻塞，直到生产者放入元素 notify 它。
具体到 Java 中，使用 BlockingQueue 接口表示阻塞队列：
public interface BlockingQueue<E> extends Queue<E> {
//添加失败时会抛出异常，比较粗暴
boolean add(E e);
//添加失败时会返回 false，比较温婉，比 add 强
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();
//移除队列中某个元素，如果存在的话返回 true，否则返回 false
boolean remove(Object o);
//检查队列中是否包含某个元素，至少包含一个就返回 true
public boolean contains(Object o);
//将当前队列所有元素移动到给定的集合中，这个方法比反复地获取元素更高效
//返回移动的元素个数
int drainTo(Collection<? super E> c);
//移动队列中至多 maxElements 个元素到指定的集合中
int drainTo(Collection<? super E> c, int maxElements);
}可以看到，在队列操作（添加/获取）当前不可用时，BlockingQueue 的方法有四种处理方式：
- 抛出异常
- 对应的是
add(),remove(),element()
- 对应的是
- 返回某个值（null 或者 false）
offer(),poll(),peek()
- 阻塞当前线程，直到操作可以进行
put(),take()
- 阻塞一段时间，超时后退出
offer,poll()
总结下来如图所示：
BlockingQueue 中不允许有 null 元素，因此在 add(), offer(), put() 时如果参数是 null，会抛出空指针。null 是用来有异常情况时做返回值的。
七种阻塞队列的前三种
Java 中提供了 7 种 BlockingQueue 的实现，在看线程池之前我根本搞不清楚究竟选择哪个，直到完整地对比总结以后，发现其实也没什么复杂。
现在我们一起来看一下这 7 种实现。
1.ArrayBlockingQueue
ArrayBlockingQueue 是一个使用数组实现的、有界的队列，一旦创建后，容量不可变。队列中的元素按 FIFO 的顺序，每次取元素从头部取，加元素加到尾部。
默认情况下 ArrayBlockingQueue 不保证线程公平的访问队列，即在队列可用时，阻塞的线程都可以争夺访问队列的资格。
不保证公平性有助于提高吞吐量。
看它的主要属性：
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
//使用数组保存的元素
final Object[] items;
//下一次取元素的索引
int takeIndex;
//下一次添加元素的索引
int putIndex;
//当前队列中元素的个数
int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
//全部操作的锁
final ReentrantLock lock;
//等待获取元素的锁
private final Condition notEmpty;
//等待添加元素的锁
private final Condition notFull;
//...
}可以看到，ArrayBlockingQueue 使用可重入锁 ReentrantLock 实现的访问公平性，两个 Condition 保证了添加和获取元素的并发控制。
构造函数：
//指定队列的容量，使用非公平锁
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();
}
//允许使用一个 Collection 来作为队列的默认元素
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock(); // Lock only for visibility, not mutual exclusion
try {
int i = 0;
try {
for (E e : c) { //遍历添加指定集合的元素
if (e == null) throw new NullPointerException();
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i; //修改 putIndex 为 c 的容量 +1
} finally {
lock.unlock();
}
}可以看到，有三种构造函数：
- 默认的构造函数只指定了队列的容量，设置为非公平的线程访问策略
- 第二种构造函数中，使用
ReentrantLock创建了 2 个Condition锁 - 第三种构造函数可以在创建队列时，将指定的元素添加到队列中
四种添加元素方法的实现
第一种 add():
public boolean add(E e) {
return super.add(e);
}
//super.add() 的实现
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}add(E) 调用了父类的方法，而父类里调用的是 offer(E)，如果返回 false 就泡出异常。
第二种 offer():
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
enqueue(e);
return true;
}
} finally {
lock.unlock();
}
}可以看到 offer(E) 方法先拿到锁，如果当前队列中元素已满，就立即返回 false，这点比 add() 友好一些；
如果没满就调用 enqueue(E) 入队：
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length) putIndex = 0;
count++;
notEmpty.signal();
}可以看到，enqueue(E) 方法会将元素添加到数组队列尾部。
如果添加元素后队列满了，就修改 putIndex 为 0 ，0.0 为啥这样，先留着回头看。
添加后调用 notEmpty.signal() 通知唤醒阻塞在获取元素的线程。
第三种 put():
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}可以看到，put() 方法可以响应中断，当队列满了，就调用 notFull.await() 阻塞等待，等有消费者获取元素后继续执行；
可以添加时还是调用 enqueue(E)。
第四种 offer(E,long,TimeUnit):
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(e);
return true;
} finally {
lock.unlock();
}
}可以看到 offer() 和 put() 方法很相似，不同之处在于允许设置等待超时时间，超过这么久如果还不能有位置，就返回 false；否则调用 enqueue(E)，然后返回 true。
总体来看添加元素很简单嘛 ✧(≖ ◡ ≖✿)嘿嘿。
四种获取元素的实现：
第一种 poll():
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}poll() 如果在队列中没有元素时会立即返回 null；如果有元素调用 dequeue():
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;
}默认情况下 dequeue() 方法会从队首移除元素（即 takeIndex 位置）。
移除后会向后移动 takeIndex，如果已经到队尾，就归零。结合前面添加元素时的归零，可以看到，其实 ArrayBlockingQueue 是个环形数组。
然后调用 itrs. elementDequeued()，这个 itrs 是 ArrayBlockingQueue 的内部类 Itrs 的对象，看起来像是个迭代器，实际上它的作用是保证循环数组迭代时的正确性，具体实现比较复杂，这里暂不介绍。
第二种 take():
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}take() 方法可以响应中断，与 poll() 不同的是，如果队列中没有数据会一直阻塞等待，直到中断或者有元素，有元素时还是调用 dequeue() 方法。
第三种 带参数的 poll():
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
return dequeue();
} finally {
lock.unlock();
}
}带参数的 poll() 方法相当于无参 poll() 和 take() 的中和版，允许阻塞一段时间，如果在阻塞一段时间还没有元素进来，就返回 null。
第四种 peek():
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
final E itemAt(int i) {
return (E) items[i];
}peel() 方法很简单，直接返回数组中队尾的元素，并不会删除元素。如果队列中没有元素返回的是 null。
一波源码看下来，ArrayBlockingQueue 使用可重入锁 ReentrantLock 控制队列的访问，两个 Condition 实现生产者-消费者模型，看起来很简单的样子，这背后要感谢 ReentrantLock 和 Condition 的功劳！
2.LinkedBlockingQueue
LinkedBlockingQueue 是一个使用链表实现的、有界阻塞队列。队列的默认最大长度为 Integer.MAX_VALUE，添加的元素按 FIFO 顺序。
LinkedBlockingQueue 属性：
//链表节点
static class Node<E> {
E item;
//对当前节点的后一个节点，有三种情况：
//1.真正的节点
//2.当前节点本身，说明当前节点是头结点 ？
//3.null，说明这个节点是最后䘝界定啊
Node<E> next;
Node(E x) { item = x; }
}
//当前容量，默认是 Integer.MAX_VALUE
private final int capacity;
//当前队列中的元素数量
private final AtomicInteger count = new AtomicInteger();
//队列中的头节点，头节点的.item 永远为 null
transient Node<E> head;
//队列的尾节点，尾节点的 next 永远为 null
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();可以看到，LinkedBlockingQueue 中有两个 ReentrantLock，一个用于添加另一个用于获取，这和 ArrayBlockingQueue 不同。
LinkedBlockingQueue 构造函数：
//使用 Integer.MAX_VALUE 作为容量
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);
}
//使用 Integer.MAX_VALUE 作为容量，同时将指定集合添加到队列中
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
final ReentrantLock putLock = this.putLock;
putLock.lock(); // Never contended, but necessary for visibility
try {
int n = 0;
for (E e : c) { //遍历添加到队列
if (e == null) //需要注意待添加集合中不能有空值
throw new NullPointerException();
if (n == capacity)
throw new IllegalStateException("Queue full");
enqueue(new Node<E>(e));
++n;
}
count.set(n);
} finally {
putLock.unlock();
}
}三种构造函数也很简单，看注释就好了。
LinkedBlockingQueue 添加元素
第一种 put(E):
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e); //构造一个节点
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) //如果之前为 0，就通知等待的
signalNotEmpty();
}可以看到，LinkedBlockingQueue 使用了 AtomicInteger 类型的 count 保存队列元素个数，在添加时，如果队列满了就阻塞等待。
这时有两种继续执行的情况：
- 有消费者取元素，
count会减少，小于队列容量 - 或者调用了
notFull.signal()
入队调用的 enqueue() 很简单，链表尾部添加节点即可:
private void enqueue(Node<E> node) {
last = last.next = node;
}在入队后，还会判断一次队列中的元素个数，如果此时小于队列容量，唤醒其他阻塞的添加线程。
最后还会判断容量，如果这时队列中没有元素，就通知 notEmpty 上阻塞的：
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}这里我很懵逼啊，为什么没有元素了要告诉取元素阻塞的线程呢？
第二种 offer(E):
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;
}可以看到 offer 是一贯的直接返回结果。
第三种 offer(E,long,TimeUnit):
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}和 ArrayBlockingQueue 一样，带阻塞时间参数的 offer() 方法会阻塞一段时间，然后没结果就返回。
LinkedBlockingQueue 获取元素
第一种 take():
public E take() throws InterruptedException {
E x;
int c = -1;
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;
}注意：这里和前面一样，都使用的是 AtomicInteger.getAndDecrement() 方法，这个方法先返回当前值，然后加 1 ，所以后面判断是判断之前的情况。
队首元素出队：
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;
}剩下的几种其实跟之前介绍的差不多，就懒得多费口舌了。
LinkedBlockingQueue 比 ArrayBlockingQueue 的优势就是添加和获取是两个不同的锁，所以并发添加/获取效率更高些，因此数组元素个数用的是 AtomicInteger 类型的，这样在添加、获取时通过判断数组元素个数可以感知到并发的获取/添加操作 ；此外就是链表比数组的优势了。
3.PriorityBlockingQueue
PriorityBlockingQueue 是基于数组的、支持优先级的、无界阻塞队列。
默认情况下队列中的元素按自然排序升序排列，我们可以实现元素的 compareTo() 指定元素的排序规则，或者在初始化它时在构造函数中传递 Comparator 排序规则。
不能保证同一优先级元素的顺序
这里就不再像前面那么详细地介绍源码了。
先看看 PriorityBlockingQueue 属性*：
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;
public PriorityBlockingQueue() {
this(DEFAULT_INITIAL_CAPACITY, null);
}
public PriorityBlockingQueue(int initialCapacity) {
this(initialCapacity, null);
}
public 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];
}PriorityBlockingQueue 在初始化时创建指定容量的数组，默认是 11 。
为什么是“无界”
这不是有界吗，为什么说是无界阻塞队列呢，答案这里：
public boolean add(E e) {
return offer(e);
}
public void put(E e) {
offer(e); // never need to block
}
public boolean offer(E e, long timeout, TimeUnit unit) {
return offer(e); // never need to block
}
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;
}可以看到，几种添加方法最终都是调用了 offer() 方法，在添加元素时，当数组中元素大于等于容量时，调用 tryGrow() 扩容：
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // grow faster if small
(oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
if (newCap > oldCap && queue == array)
newArray = new Object[newCap]; //扩容数组
} finally {
allocationSpinLock = 0;
}
}
if (newArray == null) // back off if another thread is allocating
Thread.yield();
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap); //拷贝原有数据
}
}可以看到，在扩容时，如果当前队列中元素个数小于 64 个，数组容量就就乘 2 加 2；否则变成原来的 1.5 倍（原来容量越大，扩容成本越高，所以容量设置的小一点）。
如何保证优先级
再来看看 PriorityBlockingQueue 是如何保证优先级的：
public boolean offer(E e) {
//...
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 t
}可以看到，在添加元素时，会根据队列中的定制排序 comparator 是否为空调用不同的排序方法。
不了解
Comparator和Comparable可以看这篇 Java 解惑：Comparable 和 Comparator 的区别。
如果没有设置 comparator 调用的是 siftUpComparable():
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;
}可以看到，这里的排序有点最大堆的意思，每从队尾添加一个元素都会从下往上挨个比较自己和“父节点”的大小，如果小就交换，否则就停止。
比较使用的是队列元素中重写的 compareTo() 方法。
如果设置 comparator 调用的是 siftUpUsingComparator():
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;
}同样的比较，不同的是使用的是 Comparator.compare() 方法。
了解了添加时的排序，那获取元素时是如何保证按添加的顺序取出呢？以 poll() 为例：
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return dequeue();
} finally {
lock.unlock();
}
}获取时主要调用的是 dequeue() 方法：
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;
}
}出队时同样进行了两种不同的比较，我们选其中一种 siftDownComparable() 看一下：
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;
}
}可以看到，取元素时，在移除第一个元素后，会用堆排序将当前堆再排一次序。
经过源码分析我们了解了 PriorityBlockingQueue 为什么是无界、有优先级的队列了。因为它可以扩容，在添加、删除元素后都会进行排序。
由于篇幅原因，我们将阻塞队列分两篇介绍。下一篇介绍后四种队列。