阻塞队列和ArrayBlockingQueue源码解析

什么是阻塞队列

当队列中为空时，从队列总获取元素的操作将被阻塞，当队列满时，向队列中添加元素的操作将被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，知道其它的线程往队列中插入新的元素。同样，试图往满的队列中添加新元素的线程也会被阻塞，直到有其他的线程使队列重新变的空闲起来。

处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put()	offer(e, time, unit)
移除方法	remove()	poll(e)	take()	poll(time, unit)
检查方法	element()	peek()	无	无

抛出异常：当队列满时，再向队列中插入元素，则会抛出IllegalStateException异常。当队列空时，再向队列中获取元素，则会抛出NoSuchElementException异常。
返回特殊值：当队列满时，向队列中添加元素，则返回false，否则返回true。当队列为空时，向队列中获取元素，则返回null，否则返回元素。
一直阻塞：当阻塞队列满时，如果生产者向队列中插入元素，则队列会一直阻塞当前线程，直到队列可用或响应中断退出。当阻塞队列为空时，如果消费者线程向阻塞队列中获取数据，则队列会一直阻塞当前线程，直到队列空闲或响应中断退出。
超时退出：当队列满时，如果生产线程向队列中添加元素，则队列会阻塞生产线程一段时间，超过指定的时间则退出返回false。当队列为空时，消费线程从队列中移除元素，则队列会阻塞一段时间，如果超过指定时间退出返回null。

java里的阻塞队列

ArrayBlockingQueue: 一个由数组结构组成的有界队列。此队列按照先进先出的顺序进行排序。支持公平锁和非公平锁。
LinkedBlockingQueue：一个由链表结构组成的有界队列，此队列的长度为Integer.MAX_VALUE。此队列按照先进先出的顺序进行排序。
PriorityBlockingQueue： 一个支持线程优先级排序的无界队列，默认自然序进行排序，也可以自定义实现compareTo()方法来指定元素排序规则，不能保证同优先级元素的顺序。
DelayQueue： 一个实现PriorityBlockingQueue实现延迟获取的无界队列，在创建元素时，可以指定多久才能从队列中获取当前元素。只有延时期满后才能从队列中获取元素。
SynchronousQueue： 一个不存储元素的阻塞队列，每一个put操作必须等待take操作，否则不能添加元素。支持公平锁和非公平锁。
LinkedTransferQueue： 一个由链表结构组成的无界阻塞队列，相当于其它队列，LinkedTransferQueue队列多了transfer和tryTransfer方法。
- transfer：如果当前有消费线程正在获取元素，transfer则把元素直接传给消费线程，否则加入到队列中，知道该元素被消费才返回。
- tryTransfer：如果当前有消费这正在获取元素，tryTransfer则把元素直接传给消费线程，否则立即返回false；
LinkedBlockingQueue： 一个由链表结构组成的双向阻塞队列。队列头部和尾部都可以添加和移除元素，多线程并发时，可以将锁的竞争最多降到一半。

ArrayBlockingQueue 的源码解析

ArrayBlockingQueue类的结构如下：

    public class ArrayBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable {private static final long serialVersionUID = -817911632652898426L;final Object[] items;  //用数据来存储元素的容器int takeIndex;  //下一次读取或移除的位置（remove、poll、take ）int putIndex;  //下一次存放元素的位置（add、offer、put）int count;  //队列中元素的总数final ReentrantLock lock;  //所有访问的保护锁private final Condition notEmpty;  //等待获取元素的条件private final Condition notFull;  //等待存放元素的条件略...

可以看出ArrayBlockingQueue内部使用final修饰的对象数组来存储元素，一旦初始化数组，数组的大小就不可改变。使用ReentrantLock锁来保证锁竞争，使用Condition来控制插入或获取元素时，线程是否阻塞。

  public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;//获得支持响应中断的锁lock.lockInterruptibly();try {//使用while循环来判断队列是否已满，防止假唤醒while (count == items.length)notFull.await();enqueue(e);} finally {lock.unlock();}}

首先获得锁，然后判断队列是否已满，如果已满则阻塞当前生成线程，直到队列中空闲时，被唤醒操作。队列空闲则调用enqueue 插入元素。

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条件上等待的线程 notEmpty.signal();}

把元素插入到队列中去，可以看出这个队列中的数组是环形数组结构，这样每次插入、移除的时候不需要复制移动数组中的元素。

    public E take() throws InterruptedException {final ReentrantLock lock = this.lock;//获得可响应中断锁lock.lockInterruptibly();try {//使用while循环来判断队列是否已满，防止假唤醒while (count == 0)notEmpty.await();return dequeue();} finally {lock.unlock();}}

消费者线程从阻塞队列中获取元素，如果队列中元素为空，则阻塞当前的消费者线程直到有数据时才调用dequeue方法获取元素。否则直接调用dequeue方法获取元素

    private E dequeue() {// assert lock.getHoldCount() == 1;// assert items[takeIndex] != null;final Object[] items = this.items;@SuppressWarnings("unchecked")//获取元素E x = (E) items[takeIndex];//将当前位置的元素设置为nullitems[takeIndex] = null;//这里可以看出这个数组是个环形数组if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)//修改迭代器参数itrs.elementDequeued();// 唤醒在notFull条件上等待的线程 notFull.signal();return x;}

直接从数据中获取items[takeIndex]的元素，并设置当前位置的元素为null，并设置下一次takeIndex的坐标（++takeIndex），队列元素总数-1等操作。

    public boolean offer(E e) {checkNotNull(e);final ReentrantLock lock = this.lock;//获得不可响应中断的锁lock.lock();try {if (count == items.length)return false;else {//enqueue(e);return true;}} finally {lock.unlock();}}

首先判断队列中的元素是否已满，如果已满则直接返回false，否则调用enqueue方法向队列中插入元素，插入成功返回true。

    public E poll() {final ReentrantLock lock = this.lock;//获得不可响应中断的锁lock.lock();try {return (count == 0) ? null : dequeue();} finally {lock.unlock();}}

判断队列是否为空，如果为空返回null，否则调用dequeue方法返回元素。

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

首先调用offer方法插入元素，插入成功返回true，否则抛出IllegalStateException异常。

    public E remove() {E x = poll();if (x != null)return x;elsethrow new NoSuchElementException();}

首先调用poll方法获取元素，如果不为空则直接返回，否则抛出NoSuchElementException异常。

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

首先判断队列是否已满，如果已满再循环判断超时时间是否超时，超时则直接返回false，否则阻塞该生产线程nanos时间，如果nanos时间之内唤醒则调用enqueue方法插入元素。如果队列不满则直接调用enqueue方法插入元素，并返回true。

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

首先循环判断队列是否为空，如果为空再判断是否超时，超时则返回null。不超时则等待，在nanos时间唤醒则调用dequeue方法获取元素。

    public E element() {E x = peek();if (x != null)return x;elsethrow new NoSuchElementException();}

调用peek方法获取元素，元素不为空则返回，否则抛出NoSuchElementException异常。

    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];}

调用itemAt方法获取元素。

其它的阻塞队列实现原理都类似，都是使用ReentrantLock和Condition来完成并发控制、阻塞的。

本人简书blog地址：http://www.jianshu.com/u/1f0067e24ff8
点击这里快速进入简书