源码分析-ArrayBlockingQueue

ArrayBlockingQueue

阻塞队列：

首先作为阻塞队列的第一个类，这里还是简要说明下阻塞队列。

阻塞队列都是线程安全类，可以通过put和take阻塞在满队列或者空队列上。阻塞队列不允许空元素。



java中阻塞队列有这几类：

ArrayBlockingQueue：数组构成的有界阻塞队列固定长度FIFO

DelayQueue：是一个无界的队列，只有到达一定延时之后才能被取走。FIFO

LinkedBlockingQueue：链表构成的阻塞队列可以实现任意范围长度。FIFO。吞吐量高于ArrayBlockingQueue；范围可选

LinkedTransferQueue：也是一个FIFO的无界队列，但是其提供了transfer方法，如果当前有take阻塞则直接取走，否则加入队尾

PriorityBlockingQueue：根据优先级排序的队列

SynchronousQueue：take必须等待put，反之亦然，但是效率要高于其他队列。

域

ArrayBlockingQueue是一个比较简单类，从域上就能大概判断出其同步的策略。

这里仅仅讨论主要的几个内容；

Object[] items

显然是存放元素的

final ReentrantLock lock;

一个独占的可重入锁用来同步操作

private final Condition notEmpty;

private final Condition notFull;

分别用来阻塞take和put操作。

putIndex和takeIndex显然是维护当前的存取位置；

构造器

主要说两个构造器一个是

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来实现的，通常来说公平锁意味着等待时间最长的线程会在随后更优先的获取锁，但是从实际上看这样的效率未必比非公平锁效率更高。

主要原因有两个：第一个是jvm的线程调度和操作系统的线程调度未必是不匹配，第二个是实际情况中重复性的获取锁是很常见，这样jvm会压制这种情况。更容易使得线程切换。

所以通常来说非公平锁的TPS（系统吞吐量）更高。公平锁主要是用来防止饥饿。

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) {
checkNotNull(e);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();
}
}

然后是这个带有集合的构造器。

checkNotNull是一个静态方法对于null元素会抛出异常。先获得所然后将所有元素加入数组，再设置count和put标志位。take标志位显然是0，释放锁。

加入元素的方法

阻塞队列加入元素实际上有4种方法，下面就依次介绍

offer

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

首先检查e是否为null，然后加锁，如果count的长度等于items.length则返回false，否则执行insert后返回true；然后解锁。

private void insert(E x) {
items[putIndex] = x;
putIndex = inc(putIndex);
++count;
notEmpty.signal();
}

insert用来执行存入数组，增加putIndex和count，以及唤醒notEmpty条件变量

final int inc(int i) {
return (++i == items.length) ? 0 : i;
}

inc只是用来防止溢出的。

整体思路都是正常思路，但对于条件变量还是有一些需要说明的地方。首先条件变量的出现实质是是用来替换Object.wait和Object.notify和Object.notifyAll。条件变量必须依赖某个锁生成。

条件变量通过await来阻塞，然后通过signal和signalAll来唤醒。对于非阻塞状态下的唤醒自动忽略。

add

add的调用比较有意思，首先ArrayBlockingQueue是集成自AbstractQueue的，在AbstractQueue里的add方法是这样的，实际上就是对offer方法包装一下。

public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}

而在具体的实现子类里如ArrayBlockingQueue的Add方法

public boolean add(E e) {
return super.add(e);
}

直接调用超类的方法就好，实际上由于多态的影响，会调用子类的offer方法实现后包装实现。

带时间的offer

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);
}
insert(e);
return true;
} finally {
lock.unlock();
}
}

首先检查元素非空，然后根据unit.toNanos(timeout)将时间线转换为纳秒数，然后给lock上一个可中断的锁，然后在条件变量notfull上阻塞这段时间，如果获得了所则调用insert并返回true否则返回false。

这个循环写法是java推荐的典型写法。在Condition的类中有说明。这也是显示锁比内置锁好的地方，就是策略更加灵活。

put

public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
insert(e);
} finally {
lock.unlock();
}
}

看过了带时间线的offer，put方法的实现就要简单很多了。主要就是等待notFull变量。但是需要注意的是这里必须使用while (count == items.length)防止重复唤醒。

删除元素的方法

删除元素的方法基本上就是添加的相同操作的改版。只是需要注意需要返回类型是需要强制类型转换。这里不再赘述

不过针对ArrayBlockingQueue这里还有其他的删除方式。就是remove(Object o)

public boolean remove(Object o) {
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) {
if (o.equals(items[i])) {
removeAt(i);
return true;
}
}
return false;
} finally {
lock.unlock();
}
}

这个循环的写法比较巧妙，使用了两个变量，以takeIndex为初始值，增加size次。这里删除的元素是基于equals的。找到相同的元素后调用removeAt（）操作

void removeAt(int i) {
final Object[] items = this.items;
// if removing front item, just advance
if (i == takeIndex) {
items[takeIndex] = null;
takeIndex = inc(takeIndex);
} else {
// slide over all others up through putIndex.
for (;;) {
int nexti = inc(i);
if (nexti != putIndex) {
items[i] = items[nexti];
i = nexti;
} else {
items[i] = null;
putIndex = i;
break;
}
}
}
--count;
notFull.signal();
}

如果删除的是takeIndex则删除，然后takeIndex增加1，

如果不是则移动当前i之后的所有位，使之前移，直到putIndex。

迭代器

ArrayBlockingQueue的迭代器是有一定弹性的若一致性迭代器，按照doc的介绍，其一致性主要体现在三个方面：

先预先读取一个值，然后在实际使用时在做判断。

通过最终剩余元素来保证每个值都只遍历一次

由于可能在两次next之间删除了某些元素所以有可能有null，next会跳过这些元素

对于弱主要体现在在两次next调用之间会有一些写入或者删除操作。

迭代器的域

private int remaining; // Number of elements yet to be returned
private int nextIndex; // Index of element to be returned by next
private E nextItem;    // Element to be returned by next call to next
private E lastItem;    // Element returned by last call to next
private int lastRet;   // Index of last element returned, or -1 if none

remaining表示尚未返回的元素数量。

lastItem表示上次next越过的元素。

nextIndex表示下一个应该返回的坐标。

lastItem表示下一个应该返回的元素。

lastRet表示上一个返回的元素的地址，没有则置为-1。

迭代器的构造器

Itr() {
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
lastRet = -1;
if ((remaining = count) > 0)
nextItem = itemAt(nextIndex = takeIndex);
} finally {
lock.unlock();
}
}

首先获得锁，然后是的lastRet=-1；然后是的remaining为count，然后nextIndex和nextItem，最后解锁。

迭代器的方法

迭代器只有next。hasnext和remove方法。没有其他的所以只能查看和访问不能增加。

hasNext只要剩余元素大于0就返回true；

public boolean hasNext() {
return remaining > 0;
}

next比较复杂一些。

public E next() {
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (remaining <= 0)
throw new NoSuchElementException();
lastRet = nextIndex;
E x = itemAt(nextIndex);  // check for fresher value
if (x == null) {
x = nextItem;         // we are forced to report old value
lastItem = null;      // but ensure remove fails
}
else
lastItem = x;
while (--remaining > 0 && // skip over nulls
(nextItem = itemAt(nextIndex = inc(nextIndex))) == null)
;
return x;
} finally {
lock.unlock();
}
}

首先需要进行加锁同步，然后检查remaining<=0 则抛出异常

然后让lastRet指向下一个nextIndex。

另x指向nextIndex的坐标的元素。

如果x为null则使得x为nextItem。但是让lastItem为null表示删除操作失败。

如果x不为null则使得lastItem和x指向同一个位置。

然后更新remaining和nextItem直到nextItem不为null。

最后返回x

这里需要讨论一下next的策略。

在讨论域的时候我们需要注意lastItem和nextItem的区别。迭代器为了保证一致性进行了两次取值，一次是调用前一次调用next，一次是调用next之后。所有两者可能会不一样。因为第二次有可能是null。

public void remove() {
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
int i = lastRet;
if (i == -1)
throw new IllegalStateException();
lastRet = -1;
E x = lastItem;
lastItem = null;
// only remove if item still at index
if (x != null && x == items[i]) {
boolean removingHead = (i == takeIndex);
removeAt(i);
if (!removingHead)
nextIndex = dec(nextIndex);
}
} finally {
lock.unlock();
}
}

如果lastRet的值为-1则抛出异常。使E为lastItem。然后使得lastItem为null。

如果x为null或者当前的i和X不同了，则说明当前的lastItem无效了。就不需要删除了。如果相同则说明需要删除

删除的步骤就是调用removeAt，具体上一节已经说过了，还需要考虑删除的坐标和takeIndex是否相同，如果不相同还需要使得nextIndex减一。因为删除坐标之后的所有位置都前移了一小节。

总的来看这个迭代器还是有很多问题的。不过通常来说BlockingQueue并不使用迭代器。