ArrayBlockingQueue源码解析（基于JDK8）

Posted 2021-05-23 Java与大数据进阶

@

目录

• 1 介绍
• 2 添加
  • 2.1 add
  • 2.2 offer
  • 2.3 put
  • 2.4 enqueue
• 3 删除
  • 3.1 remove
  • 3.2 poll
  • 3.3 take
  • 3.4 dequeue
• 4 其他

BlockingQueue 称为堵塞队列，可以向队列中添加元素，也可以从队列中取出元素。

当队列为空时，取出可以返回失败，抛出异常或者堵塞；当队列满时，添加可以返回失败，抛出异常或者堵塞。堵塞的方法分别是 put/take。

生产者和消费者问题可以通过堵塞队列实现，只需要在队列为空或满的时候堵塞就行了。

	Throws exception	Special value	Blocks	Times out
Insert	add(e)	offer(e)	put(e)	offer(e,timeunit)
Remove	remove()	poll()	take()	poll(time,unit)
Examine	element()	peek()	not applicable	not applicable

1 介绍

ArrayBlockingQueue 底层是数组，通过 ReentrantLock 和两个条件队列 notFull 和 notEmpty 实现添加删除以及堵塞。

基本属性包括数组 items，长度是给定的参数，无法增长。两个 index 表示接下来出队和入队的位置，也就是说当前数组中保存的元素是左闭右开的区间[takeIndex,putIndex)，注意这里的数组是循环数组，putIndex 可能小于等于 takeIndex，等于有两种情况，即该队列满或者该队列为空。

count 记录了当前元素的个数，这些元素均非null，根据 count 可以判断队列是否满或者是否空。

lock 是一个锁，通过该锁的控制和释放保证队列执行添加或删除操作时只有一个线程能进入。两个队列分别记录了堵塞的线程，等待满足条件后唤醒。

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    //保存元素的数组
    final Object[] items;
    //下一次出队的位置
    int takeIndex;
    //下一次入队的位置
    int putIndex;
    //当前非空元素的个数
    int count;
    //锁
    final ReentrantLock lock;
    //如果消费take失败，进入条件队列,
  	//当堵塞队列非空则唤醒
    private final Condition notEmpty;
    //如果生产poll失败，进入条件队列,
  	//当堵塞队列非满则唤醒
    private final Condition notFull;

    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }


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

2 添加

add 和单参数 offer 只检查一次队列是否满，put 在队列满时会堵塞，等待唤醒后重新检查。

2.1 add

add 方法使用父类的 add，由于 offer 被 ArrayBlockingQueue 重写，实际上使用 ArrayBlockingQueue 的 offer。

该方法成功为 true，失败则抛出异常。

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

2.2 offer

offer 方法在成功时返回 true，失败返回 false。

在单参数方法中，使用 lock.lock()，不会被中断，只会检查一次；在多参数方法中，使用 lock.lockInterruptibly()，会被中断并抛出异常，每次被唤醒都会重新检查，只要在设定时间之内能放入则算成功。

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 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) {
         		//直到超时，队列也没有空位
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

2.3 put

与前面的方法不同，put 在失败后会堵塞，等待其他线程的唤醒，唤醒后重新检查队列满，直到插入成功或者抛出异常。

重新检查的原因是：在其他线程执行 take 中的 dequeue 后，会唤醒一个在 notFull 中的线程，假设就是唤醒该线程，该线程会从条件队列 notFull 移动到 AQS 的同步队列最后，等待。在该线程拿到AQS锁时，在同步队列中排在它前面的线程可能也会有入队添加的操作，此时已经满了，如果不检查会出错。

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

2.4 enqueue

前面的方法都是一些检查和堵塞，真正的入队操作是 enqueue。由于前面检查过元素非 null，这里添加的 x 一定非null。

将 x 放入 putIndex 的位置，取模，并增加计数，最后，由于增加了一项，可以取出这一项，只需要唤醒一个 notEmpty 中的线程来取即可。

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)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

3 删除

remove 和单参数 poll 只检查一次队列是否满，take 在队列满时会堵塞，等待唤醒后重新检查。

3.1 remove

remove 会遍历 [takeIndex,putIndex)，由于是循环的，到达末尾会取模。在找到对应的位置 i 后，执行 removeAt 实现相应删除功能。

在 removeAt中，有两种可能，一种是删除的位置正好是 takeIndex，相当于正常的出队操作；另一种是其他位置，则需要将 removeIndex 后面的开区间(removeIndex,putIndex) 向前移动一位。最后唤醒 notFull 中的一个线程。

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;
                }
              	//相当于取模操作
                if (++i == items.length)
                    i = 0;
            } while (i != putIndex);
        }
        return false;
    } finally {
        lock.unlock();
    }
}

void removeAt(final int removeIndex) {
    // assert lock.getHoldCount() == 1;
    // assert items[removeIndex] != null;
    // assert removeIndex >= 0 && removeIndex < items.length;
    final Object[] items = this.items;
    if (removeIndex == takeIndex) {
        // removing front item; just advance
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
    } 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();
}

3.2 poll

该方法的两种重载和 offer 的两种重载基本一致。

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}
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();
    }
}

3.3 take

该方法和 put 基本一致，在执行时如果队列已空，则会堵塞。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

3.4 dequeue

类似于enqueue，该方法会唤醒 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;
}

4 其他

peek直接获取 takeIndex 处的元素，不进行出队。element 会判断是否为 null，对于 null 抛出异常。

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return itemAt(takeIndex); // null when queue is empty
    } finally {
        lock.unlock();
    }
}

final E itemAt(int i) {
    return (E) items[i];
}

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