JUC队列-ArrayBlockingQueue(一)

ArrayBlockingQueue介绍

ArrayBlockingAQueue是用数组实现的线程安全的有界的阻塞队列。

线程安全是指通过“互斥锁”保护竞争资源，实现了对线程对竞争资源的互斥访问，有界是指ArrayB咯KingQueue对应的数组是有界限的，阻塞队列是指当多个线程访问竞争资源时，当竞争资源已经被某个线程获取时，其他要获取该线程的线程需要等待。

注意：ArrayBlockingQueue不同于LinkedBlockingQueue，ArrayBlockingQueue是数组实现的，并且是有界限的；而LinkedBlockingQueue是链表实现的，是无界限的。

ArrayBlokingQueue的uml图：

说明：

ArrayBlockingQueue继承于AbstractQueue，并且它实现了BlockingQueue接口。
ArrayBlockingQueue内部是通过Object[]数组保存数据的，也就是说ArrayBlockingQueue本质上是通过数组实现的。ArrayBlockingQueue的大小，即数组的容量是创建ArrayBlockingQueue时指定的。
ArrayBlockingQueue与ReentrantLock是组合关系，ArrayBlockingQueue中包含一个ReentrantLock对象(lock)。ReentrantLock是可重入的互斥锁，ArrayBlockingQueue就是根据该互斥锁实现“多线程对竞争资源的互斥访问”。而且，ReentrantLock分为公平锁和非公平锁，关于具体使用公平锁还是非公平锁，在创建ArrayBlockingQueue时可以指定；而且，ArrayBlockingQueue默认会使用非公平锁。
ArrayBlockingQueue与Condition是组合关系，ArrayBlockingQueue中包含两个Condition对象(notEmpty和notFull)。而且，Condition又依赖于ArrayBlockingQueue而存在，通过Condition可以实现对ArrayBlockingQueue的更精确的访问。

ArrayBlockingQueue源码分析

构造方法

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

初始化数组，独占锁和两个“条件”，非空条件和满条件。

添加元素

    public void put(E e) throws InterruptedException {checkNotNull(e);//获取队列的独占锁final ReentrantLock lock = this.lock;//获取锁，如果锁处于中断状态，则抛出InterruptedException异常lock.lockInterruptibly();try {//如果队列已满，则一直等待while (count == items.length)notFull.await();//入队enqueue(e);} finally {//释放锁lock.unlock();}}

说明：put(E e)的作用是将e插入阻塞队列的尾部。如果队列已满，则等待；否则，插入元素。

在了解入队enqueue操作时，我们先了解下面几个成员的含义：

// 队列中的元素个数
int takeIndex;
// 下一个被取出元素的索引
int putIndex;
// 下一个被添加元素的索引
int count;

enqueue()的源码如下：

    private void enqueue(E x) {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;// 获取“锁”，若当前线程是中断状态，则抛出InterruptedException异常lock.lockInterruptibly();try {// 若“队列为空”，则一直等待。while (count == 0)notEmpty.await();// 取出元素return extract();} finally {// 释放“锁”lock.unlock();}
}

private E extract() {final Object[] items = this.items;// 强制将元素转换为“泛型E”E x = this.<E>cast(items[takeIndex]);// 将第takeIndex元素设为null，即删除。同时，帮助GC回收。items[takeIndex] = null;// 设置“下一个被取出元素的索引”takeIndex = inc(takeIndex);// 将“队列中元素数量”-1--count;// 唤醒notFull上的等待线程。notFull.signal();return x;
}

删除元素

这里我拿remove(Object o) 举例

  public boolean remove(Object o) {if (o == null) return false;final Object[] items = this.items;final ReentrantLock lock = this.lock;lock.lock();try {if (count > 0) {final int putIndex = this.putIndex;int i = takeIndex;do {if (o.equals(items[i])) {removeAt(i);return true;}//索引到了末尾，重置0if (++i == items.length)i = 0;//直到 i = takeIndex增长到putIndex} while (i != putIndex);}return false;} finally {lock.unlock();}}

说明：remove(Object)实际上就是对数组的进行遍历对比，只不过这个队列数组有点特殊，它是环状的数组，也就是可重用的数组，如果找到该元素，调用removeAt()

    void removeAt(final int removeIndex) {final Object[] items = this.items;//如果remove的索引时队列中的第一个元素，可以直接出队。if (removeIndex == takeIndex) {// removing front item; just advanceitems[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();//如果remove的不是第一个元素，那么直接从reomve//的那个索引开始，后面的元素全部前移一位} else {// an "interior" remove// slide over all others up through putIndex.final int putIndex = this.putIndex;for (int i = removeIndex;;) {int next = i + 1;if (next == items.length)next = 0;if (next != putIndex) {items[i] = items[next];i = next;} else {items[i] = null;this.putIndex = i;break;}}count--;if (itrs != null)itrs.removedAt(removeIndex);}notFull.signal();}