ArrayBlockingQueue你了解过吗？
概述
自java5后，jdk增加了concurrent包，concurrent中的BlockingQueue，也就是堵塞队列，BlockingQueue只是一个接口，jdk为其提供了丰富的实现类，适用于不同的场景，这篇讲的是ArrayBlockingQueue。
ArrayBlockingQueue简介
ArrayBlockingQueue继承了AbstractQueue类和实现了BlockingQueue接口，是一个基于数组的有界队列，锁是基于ReentrantLock实现，只有一个锁对象，这就导致入队/出队共用一把锁，无法实现存取并行。
应用
1.在实际应用中，设置数组大小时要充分考虑到数据量，如果设置值过小，容易造成堵塞，而且运行中无法修改大小。
2.不支持null元素，否则报NullPointerException异常。
主要方法
1.插入数据
（1）offer(E e)：如果队列没满，返回true，如果队列已满，返回false（不堵塞）。
（2）offer(E e, long timeout, TimeUnit unit)：可以设置等待时间，如果队列已满，则进行等待。超过等待时间，则返回false。
（3）put(E e)：无返回值，一直等待，直至队列空出位置。
2.获取数据
（1）poll()：如果有数据，出队，如果没有数据，返回null。
（2）poll(long timeout, TimeUnit unit)：可以设置等待时间，如果没有数据，则等待，超过等待时间，则返回null。
（3）take()：如果有数据，出队。如果没有数据，一直等待（堵塞）。
源码分析
1.AbstractQueue
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable
继承了AbstractQueue抽象类，实现了BlockingQueue接口。
2.成员变量
final Object[] items;//队列
/** 下次出队下标（take、poll、remove） */
int takeIndex;
/** 下次入队下标（pull、offer、add） */
int putIndex;
/** 队列中元素个数 */
int count;
/** 锁 */
final ReentrantLock lock;
/** 等待条件（出队） */
private final Condition notEmpty;
/** 等待条件（入队） */
private final Condition notFull;
/**
* 迭代器
*/
transient Itrs itrs = null;
从源码可以看出，ArrayBlockingQueue是基于数组的队列，锁是基于ReentrantLock。
3.构造函数
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();
}
从源码看出，实例化ArrayBlockingQueue必须要设置数组长度，这就是为什么ArrayBlockingQueue是一个有界队列
4.offer
public boolean offer(E e) {
//判断你是否为null，为null抛NullPointerException
checkNotNull(e);
//获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//如果队列已满，返回false
if (count == items.length)
return false;
else {
//入队，返回true
enqueue(e);
return true;
}
} finally {
//释放锁
lock.unlock();
}
}
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
//判断是否为null，为null抛NullPointerException
checkNotNull(e);
long nanos = unit.toNanos(timeout);
//获取锁
final ReentrantLock lock = this.lock;
//lockInterruptibly优先考虑响应中断
lock.lockInterruptibly();
try {
//如果队列已满
while (count == items.length) {
//超时直接返回false
if (nanos <= 0)
return false;
//生产线程堵塞nanos时间，也有可能被唤醒，如果超过nanos时间还未被唤醒，则nanos=0，再次循环，就会返回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;
//如果入队下标已经是最后一个，则证明队列已满，入队下标设置为0
if (++putIndex == items.length)
putIndex = 0;
count++;
//唤醒堵塞的消费线程
notEmpty.signal();
}
从源码看出，为什么offer方法设置null会抛错，因为第一句代码就是checkNotNull方法。队列已满返回false，队列未满入队，返回true。
两个offer方法的区别在于是否有等待时间，实现上基本一致。
5.put
public void put(E e) throws InterruptedException {
//判断是否为null，为null抛NullPointerException
checkNotNull(e);
//获取锁
final ReentrantLock lock = this.lock;
//lockInterruptibly优先考虑响应中断
lock.lockInterruptibly();
try {
//如果队列已满
while (count == items.length)
//一直堵塞，直至被唤醒
notFull.await();
//设置元素
enqueue(e);
} finally {
//释放锁
lock.unlock();
}
}
从源码可以看出，put和offer方法差别基本不大，只是put方法通过 notFull.await()，一直堵塞，直至被唤醒
6.poll
public E poll() {
//获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//如果队列为空，返回null，否则返回dequeue方法获取的值
return (count == 0) ? null : dequeue();
} finally {
//释放锁
lock.unlock();
}
}
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) {
//当超时后，直接返回null
if (nanos <= 0)
return null;
//将线程堵塞nanos时间，堵塞时间内可能被唤醒，超过nanos时间，nanos=0，下次循环返回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;
//如果出队是数组中最后一个，下一个出队从0开始
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
//唤醒生产的线程
notFull.signal();
return x;
}
从源码可以看出，poll方法为什么获取不到数据会返回null了，两个poll之间的区别在于是否有堵塞时间，其他基本一致
7.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十分简单，核心逻辑是队列为空，一致等待，直至被唤醒。
上一篇： L4