ArrayBlockingQueue是一个基于数组实现的有界的阻塞队列。
几个要点
- ArrayBlockingQueue是一个用数组实现的队列，所以在效率上比链表结构的LinkedBlockingQueue要快一些，但是队列长度固定，不能扩展，入列和出列使用同一把锁。LinkedBlockingQueue是入列出列两把锁，读写分离。
- 先进先出，FIFO，队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素
- 新元素插入到队列的尾部，队列检索操作则是从队列头部开始获得元素
- 利用重入锁来保证并发安全
- 初始化时必须传入容量，也就是数组的大小，不需要扩容，因为是初始化时指定容量，并循环利用数组，使用之前一定要慎重考虑好容量
- put(e)（put(e)时如果队列满了则使用notFull阻塞等待）、take()阻塞
- add(e)时如果队列满了则抛出异常
- remove()时如果队列为空则抛出异常
- offer(e)时如果队列满了则返回false
- poll()时如果队列为空则返回null
- poll(timeout, unit)时如果队列为空则阻塞等待一段时间后如果还为空就返回null
- 只使用了一个锁来控制入队出队，效率较低
定义
public class ArrayBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable
成员属性
//底层存储元素的数组。为final说明一旦初始化，容量不可变，所以是有界的。 final Object[] items;//下一个take, poll, peek or remove操作的index位置 int takeIndex;//下一个put, offer, or add操作的index位置 int putIndex;// 元素数量 int count;/** * 用于并发控制：使用经典的双Condition算法 */ final ReentrantLock lock; /** 获取操作等待条件 */ private final Condition notEmpty; /** 插入操作等待条件 */ 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);//初始化用于阻塞的ConditionnotEmpty = lock.newCondition();notFull = lock.newCondition();}
offer(E e)
往队列中添加一条元素，如果添加成功，返回true，添加失败则返回false.
在LinkedBlockingQueue中入列之后有一个自我唤醒的方法，而这里却没有，是因为LinkedBlockingQueue的入列和出列是分别不同的两把锁，读写分离。而这里读写用的是同一把锁，所以在读和写在同一时间内只能执行一个方法，就不会存在线程假死状态，但效率较慢。
public boolean offer(E e) {//元素校验 checkNotNull(e);//引用锁final ReentrantLock lock = this.lock;//上锁 lock.lock();try {//当队列塞满后，不能再继续往队列中添加元素，返回falseif (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 {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();}}
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();}}private void enqueue(E x) {final Object[] items = this.items;//入列：往数组中存入元素items[putIndex] = x;/**判断元素是否存到了数组的最后一个位置上，如果是，就把下一个元素入列的索引置为1，防止索引越界，*/if (++putIndex == items.length)putIndex = 0;//入列成功，当前队列元素数量自增count++;/**通知还在等待的出列方法，队列中已有元素，可以进行出列了*/notEmpty.signal();}
add(E e) 失败则抛异常
//调用父类的add方法 public boolean add(E e) {return super.add(e); }//调用入列方法offer，offer入列失败，抛出异常，成功则返回true public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full"); }
poll()
从队列中取出一条元素，并删除，如果取出成功，返回被取出的元素，如果取出失败，则返回null
public E poll() {final ReentrantLock lock = this.lock;//上锁 lock.lock();try {//队列为空，返回null，否则执行出列方法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 {//队列为空，超时等待，等待超时，返回nullwhile (count == 0) {if (nanos <= 0)return null;nanos = notEmpty.awaitNanos(nanos);}//执行出列方法return dequeue();} finally {//释放锁 lock.unlock();}}
E peek()
从队列中取出第一条元素，但不移除
public E peek() {final ReentrantLock lock = this.lock;lock.lock();try {return itemAt(takeIndex); // null when queue is empty} finally {lock.unlock();}}
take()
从队列中取出一条元素，如果队列为空，则线程进行无限期等待，直到有执行入列操作的线程入列成功，队列中有元素后，在入列方法中环信当前正在等待出列的线程进行出列操作。
public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();return dequeue();} finally {lock.unlock();}}private E dequeue() {// 引用队列存放的数组final Object[] items = this.items;@SuppressWarnings("unchecked")//获取要出列的元素E x = (E) items[takeIndex];//移除出列后的元素items[takeIndex] = null;/**标记下次出列的元素的索引，并判断当前出列元素是否是数组中最后一条元素，如果是则，标记下次出列元素索引为0，从数组头部开始出列*/if (++takeIndex == items.length)takeIndex = 0;//出列成功，队列长度-1count--;if (itrs != null)itrs.elementDequeued();notFull.signal();return x;}
remove()
public E remove() {E x = poll();if (x != null)return x;elsethrow new NoSuchElementException();}
- 抛异常：add、remove
- 返回特定值：offer(e)——boolean、poll()——null、peek()——null
- 阻塞：take、put
