ArrayBlockingQueue

java.util.concurrent

类 ArrayBlockingQueue<E>

java.lang.Object

java.util.AbstractCollection<E>  

java.util.AbstractQueue<E>

         java.util.concurrent.ArrayBlockingQueue<E>

 

       一个由数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部 是在队列中存在时间最长的元素。队列的尾部 是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。

       这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

       底层是通过数组来实现的一个有界阻塞队列。

       相当与一个有界缓存区，当向已满队列中存放元素会导致操作阻塞，直到队列非满

       当从已空队列中提取元素也会阻塞，直到队列非空。

---

  /** 存放元素的数组，final类型，表示数组不可变  不能扩容之类的 */

  final Object[] items;

 

  /** 下一个获取、删除元素的索引位 items index for next take, poll, peek or remove */

  int takeIndex;

 

  /** 下一个新增元素的索引位 items index for next put, offer, or add */

  int putIndex;

 

  /** 队列中元素个数 */

  int count;

  /** 读写操作锁 */

  final ReentrantLock lock;

  /** 获取元素操作的条件 */

  private final Condition notEmpty;

  /** 插入元素操作的条件 */

  private final Condition notFull;

构造函数：

public ArrayBlockingQueue(int capacity, boolean fair) {

      if (capacity <= 0)

          throw new IllegalArgumentException();

      this.items = new Object[capacity];  //初始化数组容量

      lock = new ReentrantLock(fair);  //公平锁 or 非公平锁

      notEmpty = lock.newCondition();  

      notFull =  lock.newCondition();

}

新增元素：

public boolean offer(E e) {

      checkNotNull(e);    //非空检查

      final ReentrantLock lock = this.lock;  

      lock.lock();   //上锁

      try {

          if (count == items.length)  //判断是否队列已满

              return false;

          else {    //未满

              insert(e);  //插入

              return true;

          }

      } finally {

          lock.unlock();  //释放锁

      }

}

private void insert(E x) {

      items[putIndex] = x;

      putIndex = inc(putIndex);  //判断putIndex是否已到队列尾 如果是则从0开始

      ++count;//数量+1

      notEmpty.signal();//唤醒非空条件下的线程

}

final int inc(int i) {

      return (++i == items.length) ? 0 : i;  //如果到队尾，则从0开始

}

可阻塞新增元素：

public void put(E e) throws InterruptedException {

      checkNotNull(e);

      final ReentrantLock lock = this.lock;

      lock.lockInterruptibly();  //可响应线程打断

      try {

          while (count == items.length)   //不断轮循判断  是否非满  如果满  则线程等待

              notFull.await();  //await里面让线程进入当前条件的等待队列中 并释放锁  然后会自旋(while)挂起，直到被唤醒或者超时或者CACELLED，继而获取锁

          insert(e);  //条件满足后 插入 并唤醒 删除 操作线程

      } finally {

          lock.unlock();

      }

}

移除元素：

public E poll() {

      final ReentrantLock lock = this.lock;

      lock.lock();

      try {

          return (count == 0) ? null : extract();  //如果队列为空，则返回null

      } finally {

          lock.unlock();

      }

}

private E extract() {

      final Object[] items = this.items;

      E x = this.<E>cast(items[takeIndex]);  //获取takeIndex位置元素

      items[takeIndex] = null;  //赋值为null  gc

      takeIndex = inc(takeIndex); //takeIndex自增  如果到队列尾，则从头开始

      --count;  //元素数量自减

      notFull.signal(); //唤醒非满条件线程  表示可以做新增操作

      return x;

}

可阻塞删除元素：

public E take() throws InterruptedException {

      final ReentrantLock lock = this.lock;

      lock.lockInterruptibly();  //响应线程打断

      try {

          while (count == 0)

              notEmpty.await();  //阻塞

          return extract();  //移除元素 并唤醒非满条件线程  

      } finally {

          lock.unlock();

      }

}

 

参考：http://blog.csdn.net/longeremmy/article/details/8230264

           http://www.blogjava.net/xylz/archive/2010/07/27/327265.html