引言：

java.util.concurrent.BlockingQueue阻塞队列，通常用于一个线程生产对象，并放入队列，另外一个线程获取并消费这些对象的场景，很多消息框架都有类似实现。

接口实现方法简介：

如下图所示：

入队时，因容器限制导致插入异常，使用add会抛出IllegalStateException，使用offer则直接返回false，使用put则会线程阻塞直到可以入队。

出队情况一样。

接口实现类简介：

BlockingQueue是个接口，目前有以下几个类实现了BlockingQueue接口：

ArrayBlockingQueue：

一个内部元素由数组存放的有界阻塞队列。初始化的时候必须设定容量，用先进先出(FIFO)的方式对元素进行入队和出队。

DelayQueue：

DelayQueue中内部使用的是PriorityQueue存放数据，其中的元素必须实现Delayed接口，可以判断元素的延迟是否达到，如果达到则该元素将会在DelayQueue的下一次take被调用的时候出队。

LinkedBlokingQueue：

无界阻塞队列，以 链表 结构对元素进行存储，用先进先出(FIFO)的方式对元素进行入队和出队。

PriorityBlockingQueue：

有优先级的无界阻塞队列，元素必须实现java.lang.Comparable接口，用来作为优先级的比较。因此在该队列中元素的排序取决于元素类中的实现。

SynchronousQueue：

同步队列，这是一个特殊的队列，内部一次只能容纳一个元素。如果该队列已存在一个元素，往该队列中插入一个元素的线程将会被阻塞，直到其他线程将该队列内的元素从队列中出队。

ArrayBlockingQueue源码实现：

1.属性值

final Object[] items;

int takeIndex;

int putIndex;

int count;

final Reentrant lock lock;

private final Condition notEmpty;

private final Condition notFull;

如上所示，数据存放在items这个数组，入队和出队的索性各一个，当前容量count，入队出队共享的锁lock，非空的条件Condition及非满的条件Condition。

2.入队操作(以offer为例)

步骤一，检查元素e不为null，获取超时时间，获取可重入锁。

步骤二，如果count=items.length(队列的数组已经存满)：如果超时时间为负，则失败；否则线程在。

步骤三，进入enqueue方法，添加元素，增加count值，唤醒在notEmpty这个Condition的等待队列上等待读取队列数据的线程。

public boolean offer(E e, long timeout, TimeUnit unit)

throws InterruptedException {

checkNotNull(e);

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

}

}

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

}

3.出队操作(poll为例)

步骤一，获取超时时间，获取可重入锁。

步骤二，如果count == 0(容量为空)。如果超时时间为负，则失败；notEmpty这个Condition的等待队列中 的情况，延迟时间到依然需要等待while循环结束

步骤三，进入dequeue方法，删除元素，count值，唤醒在notEmpty这个Condition的等待队列上等待将数据入队的线程。

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

}

}

private E dequeue() {

// assert lock.getHoldCount() == 1;

// assert items[takeIndex] != null;

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在一个给定的数组中，通过两个index进行循环，做到入队出队的操作。

队列为空时，使用notEmpty这个Condition让队列出队的线程阻塞等待。

当队列满时，使用notFull这个Condition让队列入队的线程阻塞等待。

入队出队使用的是同一个lock，同一时间只能做一个，因为是重入锁，持有锁的同个线程再次进入可直接进入。

Condition操作可参考本人之前结束的文章。

LinkedBlokingQueue源码实现：

1.属性介绍

static class Node<E> {

E item;

/**

* One of:

* – the real successor Node

* – this Node, meaning the successor is head.next

* – null, meaning there is no successor (this is the last node)

*/

Node<E> next;

Node(E x) { item = x; }

}

private final int capacity;

private final AtomicInteger count = new AtomicInteger();

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

如上代码所示，由Node节点构成的链表结构

head和last作为头尾，因为是transient类型，可以判断头尾不是存放具体元素的节点。

capacity是容量，为空时，容量默认是Integer.MAX_VALUE。

锁有takeLock和putLock两个，入队和出队各自处理，不相互影响。

count容量因可能有入队出队两个线程操作，使用原子操作类。

2.入队操作(以offer为例)

从如下代码可以看出，与ArrayBlockingQueue的入队操作基本类似，只有以下几点差异：

1.使用putLock防止入队操作争用，对出队操作无影响。

2.链表插入处理与数组不同

3.count原子操作类，累加方法特殊。

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;

}

private void enqueue(Node<E> node) {

last = last.next = node;

}

出队操作差不多，不再赘述。

PriorityBlockingQueue源码实现：

1.基本属性实现

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;

private PriorityQueue<E> q;

如上代码所示：

默认队列容量11

最大的队列容量为Integer.MAX_VALUE – 8

queue是队列元素存放的数组，size是当前队列中元素的个数

comparator是元素比较器，比较结果将决定元素的队列顺序

lock是出队入队的共享锁

allocationSpinLock是扩容数组分配资源时的自旋锁，CAS需要使用。

2.入队过程(以offer为例)

如下代码所示，

在入队时，判断容量是否满足需要，如果不满足，使用tryGrow扩容。

siftUpComparable是真正的插入操作方法。

而在siftUpComparable中，逻辑上使用二叉树的方式进行数据存储和获取，从而实现有优先级的获取功能。

siftUpComparable过程如下所示：

先插入到最后节点，再和父节点比较大小，小于则内容互换(结构不变，内容替换)。

入队操作

代码如下：

3.出队过程(以poll为例)

如图所示，出队的总是当前树的根节点(最小)，并取最后一个节点，与当前节点的右叶子节点比较，如果小于则互换，再与父节点比较，如果小于互换，再往父节点重复该步骤，以此方式调整树的结构。

4.数组扩容

如下代码所示，根据allocationSpinLockOffset标志进行CAS操作，判断扩容是否有线程在操作，不使用阻塞锁的方式进行并发处理。

如果扩容步骤已被其他线程抢占，则让出执行权限。

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

}

}

以上是关于阻塞队列的第一部分，DelayQueue，SynchronousQueue在后面章节中进行学习