ArrayBlockingQueue源码分析总结

本文主要是介绍ArrayBlockingQueue源码分析总结，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1，在说ArrayBlockingQueue之前，我们先说一下ReentrantLock：

ReentrantLock和Synchronized的作用差不多， java在编写多线程程序时，都是为了保证线程安全，需要对数据同步
1）相似点：
这两种同步方式有很多相似之处，它们都是加锁方式同步，而且都是阻塞式的同步，也就是说当如果一个线程获得了对象锁，进入了同步块，其他访问该同步块的线程都必须阻塞在同步块外面等待，而进行线程阻塞和唤醒的代价是比较高的（操作系统需要在用户态与内核态之间来回切换，代价很高，不过可以通过对锁优化进行改善）。
2）区别：
这两种方式最大区别就是对于Synchronized来说，它是java语言的关键字，是原生语法层面的互斥，需要jvm实现。而ReentrantLock它是JDK 1.5之后提供的API层面的互斥锁，需要lock()和unlock()方法配合try/finally语句块来完成。
3）至于Condition这个对象，当我们阅读到源码的时候，有的人可能很陌生，其实不要太对它纠结，它其实是封装了和锁相关的方法(wait(),notify(),notifyAll())，方便我们对某线程状态进行变更。

2，什么是阻塞队列

答：阻塞队列其实就是生产者-消费者模型中的容器。当生产者往队列中添加元素时，如果队列已经满了，生产者所在的线程就会阻塞，直到消费者取元素时 notify 它；消费者去队列中取元素时，如果队列中是空的，消费者所在的线程就会阻塞，直到生产者放入元素 notify 它。具体到 Java 中，使用 BlockingQueue 接口表示阻塞队列：

3，实现原理

ArrayBlockingQueue内部维护看一个数组，其内的方法别加了ReentrantLock锁来保证线程安全。

ArrayBlockingQueue是一个阻塞式的队列，继承自AbstractBlockingQueue,间接的实现了Queue接口和Collection接口。底层以数组的形式保存数据(实际上可看作一个循环数组)。常用的操作包括 add ,offer,put，remove,poll,take,peek。

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。