阻塞队列——ArrayBlockingQueue源码分析

Posted 2021-03-01 tuyang1129

一、前言

??这几天准备研究一下Java中阻塞队列的实现。Java中的阻塞队列有七种，我准备逐一研究它们的源码，然后每一个阻塞队列写一篇分析博客，这是其中的第一篇。这篇博客就来说一说阻塞队列中我认为应该是最简单的一种——ArrayBlockingQueue。

二、正文

2.1 什么是阻塞队列

??在正式分析前，先简单介绍一下什么是阻塞队列。在说阻塞队列前，先要了解生产者消费者模式：

生产者消费者模式：生产者生产产品，将生产好的产品放入一个缓冲区域，消费者消费产品，它从缓冲区域获取生产者生产的产品进行消费。缓冲区域有容量限制，若缓存区域已经满了，则生产者需要停止生产，等待缓冲区有空闲位置后，再恢复生产；若缓冲区为空，则消费者需要等待，直到缓冲区中有产品后，才能进行消费；

??阻塞队列就是基于这种模式实现的队列型容器。阻塞队列的一般实现是：我们创建队列时，指定队列的容量，当队列中元素的个数已经满时，向队列中添加元素的线程将被阻塞，直到队列不满才恢复运行，将元素添加进去；当队列为空时，向队列获取元素的线程将被阻塞，直到队列不空才恢复运行，从队列中拿出元素。

??以上是阻塞队列的一般实现，根据具体情况的不同，也会有所差异，比如有的是基于链表实现，有的是基于数组实现；有的是阻塞队列的没有容量限制（无界），而有的是有限制的（有界）。我们现在要分析的ArrayBlockingQueue就是一个基于数组实现的有界阻塞队列。下面我们就来从源码的角度分析一下ArrayBlockingQueue。

2.2 ArrayBlockingQueue类的成员变量

??我们先来了解一下ArrayBlockingQueue有哪些成员变量，知道它的成员变量对我们理解它的实现有很大的帮助：

/** 一个数组，用来存储放入队列中的元素 */
final Object[] items;

/** 此变量用来记录下一次从队列中拿出的元素，它在数组中的下标，可以理解为队列的头节点 */
int takeIndex;

/** 此变量存储下一次往队列中添加元素时，这个元素在数组中的下标，也就是记录队列尾的上一个位置 */
int putIndex;

/** 记录队列中元素的个数 */
int count;

/** 一个锁，用来保证向队列中插入、删除等操作的线程安全 */
final ReentrantLock lock;

/** 用来在队列为空时阻塞获取元素的线程，也就是用来阻塞消费者，
 * 这个变量叫notEmpty（不空），可以理解为队列空时将会被阻塞，不空时可以正常运行
 */
private final Condition notEmpty;

/** 用来在队列满时阻塞添加元素的线程，也就是用来阻塞生产者
 * 这个变量叫notFull（不满），可以理解为队列满时将会阻塞，不满时可以正常运行
 */
private final Condition notFull;

/** 遍历使用的迭代器 */
transient Itrs itrs = null;

??通过以上成员变量，我们可以得知很多信息。首先，ArrayBlockingQueue是基于数组实现的阻塞队列，由于队列是从头部获取元素，尾部添加元素，所以定义了两个变量分别记录队列头在数组中的下标，以及插入新元素时的下标。除此之外，我们可以看到，它是使用ReentrantLock来实现的线程同步，在这个lock上定义了两个Condition对象，分别用来阻塞生产者和消费者，这是生产者消费者模式非常基本的一种实现方式。

2.3 ArrayBlockingQueue的构造方法

??下面我们来看看它的构造方法：

// 仅仅指定队列容量的构造方法
public ArrayBlockingQueue(int capacity) {
    // 调用下面那个构造方法，第二个参数默认为false，表示使用非公平锁
    this(capacity, false);
}

/**
 * 此构造方法接收两个参数：
 * 1、capacity：指定阻塞队列的容量
 * 2、fair：指定创建的ReentrantLock是否是公平锁
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
    // 容量必须大于0
    if (capacity <= 0)
        throw new IllegalArgumentException();
    // 初始化存储元素的数组
    this.items = new Object[capacity];
    // 创建用于线程同步的锁lock，若fair为true，
    // 则此时创建的将是一个公平锁，反之则是非公平锁
    lock = new ReentrantLock(fair);
    // 初始化notEmpty变量，用以阻塞和唤醒消费者线程
    notEmpty = lock.newCondition();
    // 初始化notFull变量，用以阻塞生产者线程
    notFull =  lock.newCondition();
}

??上面的构造方法还是比较好理解的，唯一需要注意的地方就是用于线程同步的lock可以指定为公平锁，这也就意味着，线程的执行顺序将按时间排序，也就是先申请获取元素的线程，一定比后申请获取元素的线程，更先拿到元素，而向队列中放置元素的线程也是如此。如果我们需要这种先后顺序，可以将lock指定为公平锁，公平锁可以避免线程“饥饿”，但是公平锁比非公平锁的开销更大，因为强制要求每个线程排队，会导致阻塞和唤醒线程的次数大大增加，所以如果不是必要，最好还是使用非公平锁。

2.4 入队方法的实现

??接下来我们来看一看ArrayBlockingQueue的实现中，向队列中添加元素是如何实现的。ArrayBlockingQueue添加元素的方法有三个，分别是add，offer以及最重要的put。add和offer是Queue接口中定义的方法，任何一个实现了Queue接口的类都实现了这两个方法。但是put方法是阻塞队列才有的方法，它才是实现阻塞队列的核心方法之一。下面我们就先来分析看看put的实现：

public void put(E e) throws InterruptedException {
    // 判断元素是否为null，若为null将抛出异常
    checkNotNull(e);
    // 获取锁对象lock
    final ReentrantLock lock = this.lock;
    // 调用lock的lockInterruptibly方法加锁，lockInterruptibly可以响应中断
    // 加锁是为了防止多个线程同时操作队列，造成线程安全问题
    lock.lockInterruptibly();
    try {
        // 如果当前队列中的元素的个数为数组长度，表示队列满了，
        // 这时调用notFull.await()让当前线程阻塞，也就是让生产者阻塞
        // 而此处使用while循环而不是if，是考虑到线程被唤醒后，队列可能还是满的
        // 所以线程被唤醒后，需要再次判断，若依旧是满的，则再次阻塞
        while (count == items.length)
            notFull.await();
        
        // 调用enqueue方法将元素加入数组中
        enqueue(e);
    } finally {
        // 释放锁
        lock.unlock();
    }
}

/** 此方法将新元素加入到数组中 */
private void enqueue(E x) {
    // 获得存储元素的数组
    final Object[] items = this.items;
    // 将新元素x放入到数组中，且放入的位置就是putIndex指向的位置
    items[putIndex] = x;
    // putIndex加1，如果超过了数组的最大长度，则将其置为0，也就是数组的第一个位置
    if (++putIndex == items.length)
        putIndex = 0;
    // 元素数量+1
    count++;
    // 因为我们已经向队列中添加了元素，所以可以唤醒那些需要获取元素的线程，也就是消费者
    // 之前说过，notEmpty就是用来阻塞和唤醒消费者的
    notEmpty.signal();
}

// 判断元素是否为null
private static void checkNotNull(Object v) {
    if (v == null)
        throw new NullPointerException();
}

??以上就是ArrayBlockingQueue中put方法的实现。读了它的源码后我们可以发现，put的工作工程就是：向队列中添加一个新元素，若队列已经满了，则当前线程被阻塞，等待队列不满时被唤醒；当前线程成功添加元素后，将唤醒正在等待的消费者线程（如果有的话），消费者线程则从队列中获取元素。除了put方法外，阻塞队列还有两个方用以添加元素，就是add以及offer，这是Queue接口中定义的方法，也就是说并不是阻塞队列所特有的，所以这两个方法比较普通，我们简单地看一看即可：

public boolean offer(E e) {
    // 判断加入的元素是否为null，若为null将抛出异常
    checkNotNull(e);
    // 获取锁对象
    final ReentrantLock lock = this.lock;
    // 加锁防止线程安全问题，注意这里调用的是lock()方法，这个方法并不响应中断
    // 而之前的put方法会响应中断，以为put会阻塞，为了防止它长期阻塞，所以需要响应中断
    // 但是这个方法并不会被阻塞，所以不需要响应中断
    lock.lock();
    try {
        // 若当前队列已满，则不进行添加，直接返回false，表示添加失败
        if (count == items.length)
            return false;
        else {
            // 若队列不满，则直接调用enqueue方法添加元素，并返回true
            enqueue(e);
            return true;
        }
    } finally {
        // 解锁
        lock.unlock();
    }
}

public boolean add(E e) {
    // 调用offer方法添加元素，若offer方法返回true表示添加成功，则此方法返回true
    if (offer(e))
        return true;
    // 添加失败直接抛出异常
    else
        throw new IllegalStateException("Queue full");
}

??可以看到，这两个方法的实现比较简单。offer方法在队列满时直接放弃添加，返回false，若添加成功返回true；add方法直接调用offer方法添加元素，若添加失败，将会抛出异常。除了上面两个方法外，ArrayBlockingQueue还有一个比较特殊的方法，也是用来添加元素，并且在队列满时也会进行等待，但是并不会一直等待，而是等待指定的时间，这个方法是offer的重载方法，其代码如下：

/**
 * 此方法用来阻塞式地添加元素，但是需要指定阻塞的超时时间
 * 1、timeout：需要阻塞时间的数量级，一个long类型的整数；
 * 2、unit：用以指定时间的单位，比如TimeUnit.SECONDS表示秒，
 *  	   若timeout为10，而unit为TimeUnit.SECONDS，则表示最多阻塞100秒
 */
public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
	// 判断元素是否为null
    checkNotNull(e);
    // 获取线程需要阻塞的时间的纳秒值
    long nanos = unit.toNanos(timeout);
    // 获取锁对象
    final ReentrantLock lock = this.lock;
    // 加锁，并且lockInterruptibly方法会响应中断
    lock.lockInterruptibly();
    try {
        // 若当前队列中元素已满
        while (count == items.length) {
            // 若等待的剩余时间小于0，表示超过了等待时间，则直接返回
            if (nanos <= 0)
                return false;
            // 让当前线程等待指定的时间，使用notFull对象让线程等待一段时间
            // 方法会返回剩余的需要等待的时间
            nanos = notFull.awaitNanos(nanos);
        }
        // 调用enqueue方法将元素添加到数组中
        enqueue(e);
        // 返回true表示添加成功
        return true;
    } finally {
        // 解锁
        lock.unlock();
    }
}

2.5 出队方法的实现

??说完了向ArrayBlockingQueue中添加元素的方法，再说一说拿出队列中元素的方法。和添加元素类似，元素移出队列也有三个方法，分别是remove、poll以及阻塞队列中最关键的两个方法之一的take（另一个关键方法是put）。我们就先来看看take方法的实现：

public E take() throws InterruptedException {
    // 获取锁对象
    final ReentrantLock lock = this.lock;
    // 使用lock对象加锁，lockInterruptibly方法会响应中断
    // 目的是防止线程一直在此处阻塞，无法退出
    lock.lockInterruptibly();
    try {
        // 若当前队列中元素为0，则调用notEmpty对象的await()方法，
        // 让当前获取元素的线程阻塞，也就是阻塞消费者线程，直到被生产者线程唤醒
        while (count == 0)
            notEmpty.await();
        // 调用dequeue方法获取队投元素，并直接返回
        return dequeue();
    } finally {
        // 解锁
        lock.unlock();
    }
}

/** 此方法用来获取队头元素，同时将它从数组中删除 */
private E dequeue() {
    // 获取存储元素的数组
    final Object[] items = this.items;
    // takeIndex记录的就是队头元素的下标，使用变量x记录它
    E x = (E) items[takeIndex];
    // 将队头元素从数组中删除
    items[takeIndex] = null;
    // 队头元素删除后，原队头的下一个元素就成了新的队头，所以takeIndex + 1
    // 若takeIndex加1后超过数组的范围，则将takeIndex置为0，也就是循环使用数组空间
    // 为什么是加不是减，因为在数组中，队头在左边，队尾在右边
    if (++takeIndex == items.length)
        takeIndex = 0;
    // 元素数量-1
    count--;
    // 这里是在干嘛我也没仔细研究，好像是和队列的迭代器有关
    if (itrs != null)
        itrs.elementDequeued();
    // 当有元素出队后，队列不满，就可以被阻塞的生产者线程向队列中添加元素
    notFull.signal();
    // 返回获取到的元素值
    return x;
}

??take方法和put方法有很多的相似之处，理解了put方法，那take方法也很好理解：获得阻塞队列中队头的元素，若队列为空，则当前线程被阻塞，直到有线程向队列中添加了元素，获取成功后，将队头元素从队列中删除，然后唤醒一个被阻塞的生产者线程（如果有的话）。下面再来看看remove以及poll方法的实现：

public E poll() {
    final ReentrantLock lock = this.lock;
    // 获取元素前线加锁
    lock.lock();
    try {
        // 若队列为空，直接返回null，否则调用dequeue获取队头元素；
        return (count == 0) ? null : dequeue();
    } finally {
        // 解锁
        lock.unlock();
    }
}

// 此remove方法继承自父类
public E remove() {
    // 调用poll获取并删除队头元素
    E x = poll();
    // 若获取成功直接返回
    if (x != null)
        return x;
    // 获取失败抛出异常
    else
        throw new NoSuchElementException();
}

??这两个方法实现非常简单，就不做过多解释了。下面我们再看看另一个获取元素的方法，这个方法获取元素时，需要指定超时时间，若队列为空，则当前线程将被阻塞，但是会在指定时间后返回，代码如下：

/**
 * 方法参数timeout和unit的意义和之前指定超时时间的offer方法相同
*/
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) {
            // 若剩余等待时间小于0，则表示超时了，直接返回null
            if (nanos <= 0)
                return null;
            // 线程等待，并返回剩余等待时间
            nanos = notEmpty.awaitNanos(nanos);
        }
        // 若没有等待，或者在等待的过程中被唤醒，则调用dequeue方法获取队头元素
        return dequeue();
    } finally {
        lock.unlock();
    }
}

2.6 ArrayBlockingQueue的优缺点

??对于ArrayBlockingQueue的源码的阅读就止步于此，它的实现比较简单，看完上面的代码并理解，我认为就足够了。下面我们来讨论讨论它的优缺点。

??首先是优点，我个人认为ArrayBlockingQueue没有什么比较明显的优点，除了实现简单。再说说缺点，那就比较明显了：

• ArrayBlockingQueue只有一把锁，无论是添加还是获取元素使用的都是同一个锁对象，这也就导致了添加和获取不能同时执行，所以性能低下。但是，实际情况下，添加元素操作的是队尾，而获取元素操作的是队头，它们之间发生线程冲突的概率比较小，所以使用一把锁并不是一种好的实现方式。
• ArrayBlockingQueue基于数组实现，数组并不适用于随机删除元素，因为如果删除数组中间的元素，则这之后的元素都需要向前移动一个位置。而ArrayBlockingQueue支持remove（Object o）方法，删除指定元素。当然，这严格来讲并不是一个缺点，毕竟队列就是尾进头出，随机删除元素的操作虽然支持，但是一般不使用。

三、总结

??关于ArrayBlockingQueue的内容就说到这里。这种阻塞队列的实现较为简单，只要理解了take和put方法，基本上就足够了。但是，正因为它的实现比较简单，所以性能上并不是太好，毕竟内部只使用一把锁，所以这种阻塞队列用的也不是特别多。

四、参考

• https://mp.weixin.qq.com/s/EN7qY1w4e8C0ZXiP7i82TA