27. 并发终结之ArrayBlockingQueue

我们再来看ArrayBlockingQueue的源码，它的底层是基于Object[]数组的（数组是内存中一段连续的空间，所以插入数据会比较方便），所以肯定是有界队列。使用了一个ReentrantLock来保证线程安全（粒度较粗，会带来更多锁的竞争，不利于高并发）。

内部成员变量

items是object类型的数组。
putIndex是压入元素的指针。
takeIndex是取出元素的指针。
cout是queue里面元素的总和。
后面queue是不是满了，通过count和items.length二者相比较来判断的。
如果一直put元素，没有任何take元素的操作，那么++putIndex=item.length的时候putIndex会被重置为0，此时count==item.length，所以put操作会被阻塞。
同理如果put满了，take操作开始拿空了queue，那么++takeIndex=item.length的时候，count=0，takeIndex会被重置为0。

/** The queued items */
final Object[] items;
/** items index for next take, poll, peek or remove */
int takeIndex;
/** items index for next put, offer, or add */
int putIndex;
/** Number of elements in the queue */
int count;

/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook.
 */
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;

add

add方法是queue满的时候插入不进去会抛出IllegalStateException("Queue full")的异常，且add方法其实是调用的offer()方法，offer返回true表示add成功，否则抛出异常。

public boolean add(E e) {
    return super.add(e);
}
public boolean add(E e) {
    //add的本质是调用offer方法
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

offer

offer方法是返回boolean值的

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //如果count等于items数组的长度，表示queue已经满了，返回false
        if (count == items.length)
            return false;
        else {
            //调用enqueue()方法添加元素到queue
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    //将元素放到putIndex的位置，然后putIndex往后移一位
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
     //当有元素插入，则唤醒notEmpty等待队列
    notEmpty.signal();
}   

put

put在queue满的情况下，再添加元素会调用notFull.await()进入阻塞，等待take()方法取出元素后notFull.signal()。
且put使用的是lock.lockInterruptibly()是会响应中断的，即抛出InterruptedException

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    //lock过程中响应中断
    lock.lockInterruptibly();
    try {
        //是while循环，避免过早唤醒以及虚假唤醒
        while (count == items.length)
            //如果queue满了，就阻塞
            notFull.await();
        //唤醒之后再将元素加入到queue里
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

offer(time)

offer(time)是带返回值的，但内部调用的是awaitNanos(time)，所以会阻塞一定时间，即等待一段时间后自己醒来，如果queue还是满的，就返回false；如果不是满的，就将元素加入queue，并返回true。

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) {
            //如果被唤醒之后发现queue还是满的，且等待时间为0，就直接return false
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        //否则就将新元素加入queue，并返回true。
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

remove

remove和add是一对的，所以实现也是类似，调用的是poll()方法，如果poll的返回值不是null，则返回该值；如果是null，则抛出NoSuchElementException。

public E remove() {
    E x = poll();
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

poll

poll方法有值就返回值（从queue里移除），没有值就返回null。

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}
//dequeue主要还是对items数组进行操作，最后唤醒notFull的等待队列
private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return x;
}

peek

peek操作是返回queue里的头元素，即items数组里takeIndex位置的元素。

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return itemAt(takeIndex); // null when queue is empty
    } finally {
        lock.unlock();
    }
}
//返回items数组里takeIndex的值
final E itemAt(int i) {
    return (E) items[i];
}

take

take对应的是put操作，take在queue为空的时候，会阻塞，也是会响应中断的。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            //在循环中判断count是不是为0，如果是0，表示queue为空，notEmpty.await进入等待
            notEmpty.await();
        //唤醒之后就返回当前takeindex的值
        return dequeue();
    } finally {
        lock.unlock();
    }
}

poll(time)

poll(time)与offer(time)相对应，poll一段时间，如果queue还是为空就返回null值，否则返回takeIndex对应的数组里的值。

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}