并发编程之阻塞队列ArrayBlockingQueue

一. ArrayBlockingQueue类图结构

ArrayBlockingQueue是一个有界阻塞队列，先看一下ArrayBlockingQueue的几个重要元素

/**数组存放数据*/

final Object[] items;

/** 记录取数据的位置，可以看成是队头 */

int takeIndex;

/** 记录存数据的位置，可以看成是队尾 */

int putIndex;

/** 队列长度，元素的个数 */

int count;

/** 锁 */

final ReentrantLock lock;

/** 未空条件，当队列是空的时候，调用notEmpty的await方法，让取元素的线程等待，当存放元素的时候，调用notEmpty的single方法，唤醒取元素的等待线程 */

private final Condition notEmpty;

/** 未满条件，当队列是满的时候，调用notFull的await方法，让存元素的线程等待，当取元素的时候，调用notFull的single方法，唤醒存元素的等待线程 */

private final Condition notFull;

构造方法

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的方法

offer(E e)方法

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

}

}

private void enqueue(E x) {

final Object[] items = this.items;

items[putIndex] = x;

if (++putIndex == items.length)

putIndex = 0;

count++;

notEmpty.signal();

}

可以看出offer方法的执行流程就是：判断所加元素是否为空，为空就抛出空指针异常，然后对lock加锁，保证执行offer方法的安全性，由于ArrayBlockingQueue的其他存取元素的方法都用的是同一个锁，所以导致当有线程调用其他存取方法的时候，也会因为获取不到锁而处于阻塞状态。获取到锁后，先判断队列是否已经满了，如果已经满了，在返回之前先解锁，再返回false，如果没有满，就加入到数组中，将putIndex后移一位，可以看成是将队尾后移一位。如果当putIndex等于最大长度的时候，说明队列已经满了，然后将putIndex置为0。这个操作相当于把ArrayBlockingQueue置为了一个循环队列，当队列满了后，再存放元素就会将元素放到队头，也就是数组下标是0的位置。count加1，然后解锁非空的限制，因为现在队列刚存放了一个元素，队列里面有值了，唤醒那些在队列是空的时候去取元素的线程可以过来取元素了。返回true，最后解锁。

上面说的这个过程有点乱，总结一下offer方法

1.空指针判断

2.获取锁

3.队列满了直接返回false

4.队列没有满存放元素，唤醒因为队列是空而陷入等待的取元素线程，返回true

5.解锁

在这个过程的中，调用offer的方法是有可能因为竞争lock锁而处于等待状态的，但是它不会由于队列是满了的导致放不了元素而一直处在等待状态，要是队列是满的话，它直接就返回false了。

boolean offer(E e, long timeout, TimeUnit unit)方法

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

}

}

这个方法的意思当存放元素的时候，如果队列满了，就等待timeout的时间，等待完这段时间要是队列还是满的，存放不了元素，就返回false。

利用一个while循环来判断队列是否是满的，是满的，则陷入等待中，如果在规定时间之内被唤醒了存放元素，返回true，如果在规定时间之内没有被唤醒就返回false。

void put(E e)

public void put(E e) throws InterruptedException {

checkNotNull(e);

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

while (count == items.length)

notFull.await();

enqueue(e);

} finally {

lock.unlock();

}

}

put方法也是存放元素，执行流程如下：

1.空指针异常判断

2.获取锁

3.循环判断队列是否已经放满，如果是满的就被阻塞，陷入等待中，等待取元素时notFull条件的唤醒。

4.存放元素，唤醒notEmpty陷入等待的取线程

5.释放锁

office和put方法的区别

1.office方法有返回值，存入成功返回true，失败返回false，put方法无返回值。

2.office方法在获取锁后无法被中断，take方法会被中断

3.office方法在获取到锁后，存放元素不会被阻塞，take方法存放元素有可能会被阻塞。

poll()方法

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return (count == 0) ? null : 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;

}

poll方法是用于从队列中获取元素，如果取不到就返回null。poll的执行流程是：

1.先获取锁

2.判断队列的元素数量是否为0，如果为0返回null。

3.如果队列元素的数量不为0，则从队列中获取到元素，takeIndex++（也就是队头往后移动一位），然后元素的个数减一，然后唤醒插入的时候因为队列是满的而等待的线程。

4.释放锁

E poll(long timeout, TimeUnit unit)

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时间内被唤醒的，那么nanos这个返回值会大于0

nanos = notEmpty.awaitNanos(nanos);

}

return dequeue();

} finally {

lock.unlock();

}

}

poll带有等待时间，如果在timeout时间之内被唤醒，可以重新尝试取值。

E 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方法是从队列的头部中取回返回值，也是先获取锁，然后进入一个队列是空的while循环，要是队列是空就会一直陷入被阻塞住，等待被唤醒，唤醒后会再跑一次while循环，进行再判断一下，不为空就会去队列中取值，然后解锁。

E peek()方法

public E peek() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return itemAt(takeIndex); // null when queue is empty

} finally {

lock.unlock();

}

}

peek方法只是展现队头的值，而不是取出对头的值

从以上方法不难发现，无论是取数据还是存数据都会先加锁，然后再操作，如果想要有等待效果的话，就可以使用condition的wait方法，让当前操作的线程陷入等待状态。

poll方法和take方法的区别

1.poll方法要是取不到元素的话会返回nll，take方法要是取不到元素的话就陷入等待。

2.poll方法无法被中断，take方法可以被中断。

读源码的一点感受

就是condition的wait方法和while结合使用，当满足条件要将当前线程挂起的时候，不是用if来判断，而是用while来判断的，这说明当被唤醒后还需要再判断一次？为什么再获取到锁的情况一下，还需要再判断一次？现在还没有理解这样的一个操作。

三.阻塞队列的总结

阻塞队列的用法

1.消费者生产者模式，利用阻塞队列来实现就很简单，只需有消费者和生产者线程就够了，里面的取和存都由阻塞队列控制就好了。如果不让用阻塞队列的话，其实只需要拿出阻塞队列的源码来操作就好了。

2.线程池，线程池中用阻塞队列存放任务。

阻塞队列方法对比总结

add 和remove方法抛出异常：这时候插入和取出在不能立即被执行的时候就会抛出异常。

offer和poll方法特殊值：插入和取出在不能被立即执行的情况下会返回一个特殊的值（true 或者 false）

put和take方法阻塞：插入和取出操作在不能被立即执行时会阻塞线程，直到条件成熟，被其他线程唤醒

offer(long timeout, TimeUnit unit)和poll(long timeout, TimeUnit unit)超时：插入和取出操作在不能立即执行的时候会被阻塞一定的时候，如果在指定的时间内没有被执行，那么会返回一个特殊值。

不同阻塞队列的总结

ArrayBlockingQueue和LinkedBlockingQueue是最为常用的阻塞队列，前者使用一个有边界的数组来作为存储介质，而后者使用了一个没有边界的链表来存储数据。

PriorityBlockingQueue是一个优先阻塞队列。所谓优先队列，就是每次从队队列里面获取到的都是队列中优先级最高的，对于优先级，PriorityBlockingQueue需要你为插入其中的元素类型提供一个Comparator，PriorityBlockingQueue使用这个Comparator来确定元素之间的优先级关系。底层的数据结构是堆，也就是我们数据结构中的那个堆。

DelayQueue是一个延时队列，所谓延时队列就是消费线程将会延时一段时间来消费元素。

SynchronousQueue是最为复杂的阻塞队列。SynchronousQueue和前面分析的阻塞队列都不同，因为SynchronousQueue不存在容量的说法，任何插入操作都需要等待其他线程来消费，否则就会阻塞等待，看到这种队列心里面估计就立马能联想到生产者消费者的这种模式了，没错，就可以使用这个队列来实现。