public class ArrayBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable {// 版本序列号private static final long serialVersionUID = -817911632652898426L;// 存放实际元素的数组final Object[] items;// 取元素索引int takeIndex;// 获取元素索引int putIndex;// 队列中的项int count;// 可重入锁final ReentrantLock lock;// 等待获取条件private final Condition notEmpty;// 等待存放条件private final Condition notFull;// 迭代器transient Itrs itrs = null;
}

    public ArrayBlockingQueue(int capacity) {// 调用两个参数的构造函数this(capacity, false);}

    public ArrayBlockingQueue(int capacity, boolean fair) {// 初始容量必须大于0if (capacity <= 0)throw new IllegalArgumentException();// 初始化数组this.items = new Object[capacity];// 初始化可重入锁lock = new ReentrantLock(fair);// 初始化等待条件notEmpty = lock.newCondition();notFull =  lock.newCondition();}

    public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) {// 调用两个参数的构造函数this(capacity, fair);// 可重入锁final ReentrantLock lock = this.lock;// 上锁lock.lock(); // Lock only for visibility, not mutual exclusiontry {int i = 0;try {for (E e : c) { // 遍历集合// 检查元素是否为空
                    checkNotNull(e);// 存入ArrayBlockingQueue中items[i++] = e;}} catch (ArrayIndexOutOfBoundsException ex) { // 当初始化容量小于传入集合的大小时，会抛出异常throw new IllegalArgumentException();}// 元素数量count = i;// 初始化存元素的索引putIndex = (i == capacity) ? 0 : i;} finally {// 释放锁
            lock.unlock();}}

    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) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;// 获取数组final Object[] items = this.items;// 将元素放入items[putIndex] = x;if (++putIndex == items.length) // 放入后存元素的索引等于数组长度（表示已满）// 重置存索引为0putIndex = 0;// 元素数量加1count++;// 唤醒在notEmpty条件上等待的线程
        notEmpty.signal();}

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

    public E take() throws InterruptedException {// 可重入锁final ReentrantLock lock = this.lock;// 如果当前线程未被中断，则获取锁，中断会抛出异常
        lock.lockInterruptibly();try {while (count == 0) // 元素数量为0，即Object数组为空// 则等待notEmpty条件
                notEmpty.await();// 出队列return 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];// 该索引的值赋值为nullitems[takeIndex] = null;// 取值索引等于数组长度if (++takeIndex == items.length)// 重新赋值取值索引takeIndex = 0;// 元素个数减1count--;if (itrs != null) itrs.elementDequeued();// 唤醒在notFull条件上等待的线程
        notFull.signal();return x;}

    public E poll() {// 重入锁final ReentrantLock lock = this.lock;// 获取锁
        lock.lock();try {// 若元素个数为0则返回null，否则，调用dequeue，出队列return (count == 0) ? null : dequeue();} finally {// 释放锁
            lock.unlock();}}

    public void clear() {// 数组final Object[] items = this.items;// 可重入锁final ReentrantLock lock = this.lock;// 获取锁
        lock.lock();try {// 保存元素个数int k = count;if (k > 0) { // 元素个数大于0// 存数元素索引final int putIndex = this.putIndex;// 取元素索引int i = takeIndex;do {// 赋值为nullitems[i] = null;if (++i == items.length) // 重新赋值ii = 0;} while (i != putIndex);// 重新赋值取元素索引takeIndex = putIndex;// 元素个数为0count = 0;if (itrs != null)itrs.queueIsEmpty();for (; k > 0 && lock.hasWaiters(notFull); k--) // 若有等待notFull条件的线程，则逐一唤醒
                    notFull.signal();}} finally {// 释放锁
            lock.unlock();}}

package com.hust.grid.leesf.collections;import java.util.concurrent.ArrayBlockingQueue;class PutThread extends Thread {private ArrayBlockingQueue<Integer> abq;public PutThread(ArrayBlockingQueue<Integer> abq) {this.abq = abq;}public void run() {for (int i = 0; i < 10; i++) {try {System.out.println("put " + i);abq.put(i);Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}}
}class GetThread extends Thread {private ArrayBlockingQueue<Integer> abq;public GetThread(ArrayBlockingQueue<Integer> abq) {this.abq = abq;}public void run() {for (int i = 0; i < 10; i++) {try {System.out.println("take " + abq.take());Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}}
}public class ArrayBlockingQueueDemo {public static void main(String[] args) {ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<Integer>(10);PutThread p1 = new PutThread(abq);GetThread g1 = new GetThread(abq);p1.start();g1.start();}
}

put 0
take 0
put 1
take 1
put 2
take 2
put 3
take 3
put 4
take 4
put 5
take 5
put 6
take 6
put 7
take 7
put 8
take 8
put 9
take 9

put 0
take 0
put 1
take 1
put 2
take 2
put 3
take 3
put 4
take 4
put 5
take 5
put 6
take 6
put 7
take 7
put 8
take 8
put 9
take 9