Java源码解析——集合框架（二）——ArrayBlocki…

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, Serializable

private final E[] items;//底层数据结构

private int takeIndex;//用来为下一个take/poll/remove的索引（出队）

private int putIndex;//用来为下一个put/offer/add的索引（入队）

private int count;//队列中元素的个数

private final ReentrantLock lock;//锁

private final Condition notEmpty;//等待出队的条件

private final Condition notFull;//等待入队的条件

/**
* 创造一个队列，指定队列容量，指定模式
* @param fair
* true：先来的线程先操作
* false：顺序随机
*/
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = (E[]) new Object[capacity];//初始化类变量数组items
    lock = new ReentrantLock(fair);//初始化类变量锁lock
    notEmpty = lock.newCondition();//初始化类变量notEmpty Condition
    notFull = lock.newCondition();//初始化类变量notFull Condition
}

/**
* 创造一个队列，指定队列容量，默认模式为非公平模式
* @param capacity <1会抛异常
*/
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)//数组满了
                return false;
            else {//数组没满
                insert(e);//插入一个元素
                return true;
            }
        } finally {
            lock.unlock();
        }
}

/**
* 在队尾插入一个元素,并设置了超时等待的时间
* 如果数组已满，则进入等待，直到出现以下三种情况：
* 1、被唤醒
* 2、等待时间超时
* 3、当前线程被中断
*/
public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {
　　if (e == null)
　　　　throw new NullPointerException();
　　long nanos = unit.toNanos(timeout);//将超时时间转换为纳秒
　　final ReentrantLock lock = this.lock;
        /*
         * lockInterruptibly():
         * 1、 在当前线程没有被中断的情况下获取锁。
         * 2、如果获取成功，方法结束。
         * 3、如果锁无法获取，当前线程被阻塞，直到下面情况发生：
         * 1）当前线程(被唤醒后)成功获取锁
         * 2）当前线程被其他线程中断
         * 
         * lock()
         * 获取锁，如果锁无法获取，当前线程被阻塞，直到锁可以获取并获取成功为止。
         */
　　lock.lockInterruptibly();//加可中断的锁
　　try {
　　　　for (;;) {
　　　　　　if (count != items.length) {//队列未满
　　　　　　　　insert(e);
　　　　　　　　return true;
　　　　　　}
　　　　　　if (nanos <= 0)//已超时
　　　　　　　　return false;
　　　　　　try {
　　　　　　　　/*
　　　　　　　　* 进行等待：
　　　　　　　　* 在这个过程中可能发生三件事：
　　　　　　　　* 1、被唤醒-->继续当前这个for(;;)循环
　　　　　　　　* 2、超时-->继续当前这个for(;;)循环
　　　　　　　　* 3、被中断-->之后直接执行catch部分的代码
　　　　　　　　*/
　　　　　　　　nanos = notFull.awaitNanos(nanos);//进行等待（在此过程中，时间会流失,在此过程中，线程也可能被唤醒）
　　　　　　} catch (InterruptedException ie) {//在等待的过程中线程被中断
　　　　　　　　notFull.signal(); // 唤醒其他未被中断的线程
　　　　　　　　throw ie;
　　　　　　}
　　　　}
　　} finally {
　　　　lock.unlock();
　　}
}

private void insert(E x) {
    items[putIndex] = x; // 元素添加到数组里
    putIndex = inc(putIndex); // 放数据索引+1，当索引满了变成0
    ++count; // 元素个数+1
    notEmpty.signal(); // 使用条件对象notEmpty通知，比如使用take方法的时候队列里没有数据，被阻塞。这个时候队列insert了一条数据，需要调用signal进行通知
}

final int inc(int i) {
    return (++i == items.length) ? 0 : I;
}

/**
* 在队尾插入一个元素
* 如果队列满了，一直阻塞，直到数组不满了或者线程被中断
*/
public void put(E e) throws InterruptedException {
　　if (e == null)
　　　　throw new NullPointerException();
　　final E[] items = this.items;
　　final ReentrantLock lock = this.lock;
　　lock.lockInterruptibly();
　　try {
　　　　try {
　　　　　　while (count == items.length)//队列满了，一直阻塞在这里
　　　　　　　　/*
　　　　　　　　* 一直等待条件notFull，即被其他线程唤醒
　　　　　　　　* （唤醒其实就是，有线程将一个元素出队了，然后调用notFull.signal()唤醒其他等待这个条件的线程，同时队列也不慢了）
  　　　　　　　*/
　　　　　　　　notFull.await();
　　　　} catch (InterruptedException ie) {//如果被中断
　　　　　　notFull.signal(); // 唤醒其他等待该条件（notFull，即入队）的线程
　　　　　　throw ie;
　　　　}
　　　　insert(e);
　　} finally {
　　　　lock.unlock();
　　}
}

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 加锁，保证调用poll方法的时候只有1个线程
    try {
        return (count == 0) ? null : extract(); // 如果队列里没元素了，返回null，否则调用extract方法
    } finally {
        lock.unlock(); // 释放锁，让其他线程可以调用poll方法
    }
}

private E extract() {
　　final E[] items = this.items;
　　E x = items[takeIndex];//获取出队元素
　　items[takeIndex] = null;//将出队元素位置置空
　　/*
　　* 第一次出队的元素takeIndex==0,第二次出队的元素takeIndex==1
　　* (注意：这里出队之后，并没有将后面的数组元素向前移)
　　*/
　　takeIndex = inc(takeIndex);
　　--count;//数组元素个数-1
　　notFull.signal();//数组已经不满了，唤醒其他等待notFull条件的线程
　　return x;//返回出队的元素
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 加锁，保证调用take方法的时候只有1个线程
    try {
        while (count == 0) // 如果队列空，阻塞当前线程，并加入到条件对象notEmpty的等待队列里
            notEmpty.await(); // 线程阻塞并被挂起，同时释放锁
        return extract(); // 调用extract方法
    } finally {
        lock.unlock(); // 释放锁，让其他线程可以调用take方法
    }
}

public boolean remove(Object o) {
    if (o == null) return false;
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock(); // 加锁，保证调用remove方法的时候只有1个线程
    try {
        for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) { // 遍历元素
            if (o.equals(items[i])) { // 两个对象相等的话
                removeAt(i); // 调用removeAt方法
                return true; // 删除成功，返回true
            }
        }
        return false; // 删除成功，返回false
    } finally {
        lock.unlock(); // 释放锁，让其他线程可以调用remove方法
    }
}

void removeAt(int i) {
    final Object[] items = this.items;
    if (i == takeIndex) { // 如果要删除数据的索引是取索引位置，直接删除取索引位置上的数据，然后取索引+1即可
        items[takeIndex] = null;
        takeIndex = inc(takeIndex);
    } else { // 如果要删除数据的索引不是取索引位置，移动元素元素，更新取索引和放索引的值
        for (;;) {
            int nexti = inc(i);
            if (nexti != putIndex) {
                items[i] = items[nexti];
                i = nexti;
            } else {
                items[i] = null;
                putIndex = i;
                break;
            }
        }
    }
    --count; // 元素个数-1
    notFull.signal(); // 使用条件对象notFull通知，比如使用put方法放数据的时候队列已满，被阻塞。这个时候消费了一条数据，队列没满了，就需要调用signal进行通知 
}