阻塞队列BlockingQueue之ArrayBlockingQueue

Posted 2021-03-15 yrjns

 

ArrayBlockingQueue  是数组实现的有界阻塞队列，此队列按照先进先出（FIFO）的原则对元素进行排序。

构造方法：

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();//初始化非满等待队列
    }

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 exclusion
        try {
            int i = 0;
            try {
                for (E e : c) {
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally {
            lock.unlock();
        }
    }

 final Object[] items  构造以后，大致是如下的数组结构：

 

 

Add 方法

以 add 方法作为入口，在 add 方法中会调用父类的 add 方法，也就是 AbstractQueue.如果看源码看得比较多的话，一般这种写法都是调用父类的模版方法来解决通用性问题

public boolean add(E e) {
 return super.add(e);
}

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

从父类的 add 方法可以看到，这里做了一个队列是否满了的判断，如果队列满了直接抛出一个异常

 

offer 方法

add 方法最终还是调用 offer 方法来添加数据，返回一个添加成功或者失败的布尔值返回

public boolean offer(E e) {
 checkNotNull(e); //1、对请求数据做判断
 final ReentrantLock lock = this.lock;
 lock.lock();//2 添加重入锁
 try {
   f (count == items.length)
     return false;//  3  判断队列长度，如果队列长度等于数组长度，表示满了直接返回 false
   else {
     enqueue(e);// 4 否则，直接调用 enqueue 将元素添加到队列中
     return true;
   }
 } finally {
   lock.unlock();
 }
}

 

enqueue

这个是最核心的逻辑，方法内部通过 putIndex 索引直接将元素添加到数组 items

private void enqueue(E x) {
 // assert lock.getHoldCount() == 1;
 // assert items[putIndex] == null;
 final Object[] items = this.items;
 items[putIndex] = x; //通过 putIndex 对数据赋值
 if (++putIndex == items.length) // 当putIndex 等于数组长度时，将 putIndex 重置为 0
   putIndex = 0;
 count++;//记录队列元素的个数
 notEmpty.signal();//唤醒处于等待状态下的线程，表示当前队列中的元素不为空,如果存在消费者线程阻塞，就可以开始取出元素
}

putIndex 为什么会在等于数组长度的时候重新设置为 0?

因为 ArrayBlockingQueue 是一个 FIFO 的队列，队列添加元素时，是从队尾获取 putIndex 来存储元素，当 putIndex等于数组长度时，下次就需要从数组头部开始添加了

下面这个图模拟了添加到不同长度的元素时，putIndex 的变化，当 putIndex 等于数组长度时，不可能让 putIndex 继续累加，否则会超出数组初始化的容量大小。同时大家还需要思考两个问题:

 

 

put 方法

put 方法和 add 方法功能一样，差异是 put 方法如果队列满了，会阻塞。

public void put(E e) throws InterruptedException {
 checkNotNull(e);
 final ReentrantLock lock = this.lock;
 lock.lockInterruptibly(); //这个也是获得锁，但是和 lock 的区别是，这个方法优先允许在等待时由其他线程调用等待线程的 interrupt 方法来中断等待直接返回。而 lock方法是尝试获得锁成功后才响应中断
 try {
   while (count == items.length)
     notFull.await();//队列满了的情况下，当前线程将会被 notFull 条件对象挂起加到等待队列中
     enqueue(e);
 } finally {
   lock.unlock();
 }
}

 

 

take 方法

take 方法是一种阻塞获取队列中元素的方法
它的实现原理很简单，有就删除没有就阻塞，注意这个阻塞是可以中断的，如果队列没有数据那么就加入 notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put 线程添加了数据，那么 put 操作将会唤醒 take 线程，执行 take 操作。

public E take() throws InterruptedException {
 final ReentrantLock lock = this.lock;
 lock.lockInterruptibly();
 try {
   while (count == 0)
   notEmpty.await(); //如果队列为空的情况下，直接通过 await 方法阻塞
   return dequeue();
 } finally {
   lock.unlock();
 }
}

 

 

 

如果队列中添加了元素，那么这个时候，会在 enqueue 中调用 notempty.signal 唤醒 take 线程来获得元素

 

 

dequeue 方法

这个是出队列的方法，主要是删除队列头部的元素并发返回给客户端
takeIndex，是用来记录拿数据的索引值

private E dequeue() {
 // assert lock.getHoldCount() == 1;
 // assert items[takeIndex] != null;
 final Object[] items = this.items;
 @SuppressWarnings("unchecked")
 E x = (E) items[takeIndex]; //默认获取 0 位置的元素
 items[takeIndex] = null;//将该位置的元素设置为空
 if (++takeIndex == items.length)//这里的作用也是一样，如果拿到数组的最大值，那么重置为 0，继续从头部位置开始获取数据
   takeIndex = 0;
 count--;//记录 元素个数递减
 if (itrs != null)
   itrs.elementDequeued();//同时更新迭代器中的元素数据
 notFull.signal();//触发 因为队列满了以后导致的被阻塞的线程
 return x;
}

 

itrs.elementDequeued()

ArrayBlockingQueue 中，实现了迭代器的功能，也就是可以通过迭代器来遍历阻塞队列中的元素
所以 itrs.elementDequeued() 是用来更新迭代器中的元素数据的
takeIndex 的索引变化图如下，同时随着数据的移除，会唤醒处于 put 阻塞状态下的线程来继续添加数据

 

 

remove 方法

remove 方法是移除一个指定元素

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])) {//从takeIndex 下标开始，找到要被删除的元素
       removeAt(i);//移除指定元素
       return true;//返回执行结果
     }
    //当前删除索引执行加 1 后判断是否与数组长度相等
    //若为 true，说明索引已到数组尽头，将 i 设置为 0
     if (++i == items.length)
       i = 0;
     } while (i != putIndex);//继续查找，直到找到最后一个元素
   }
   return false;
 } finally {
   lock.unlock();
 }
}

 

也可以参考：https://www.cnblogs.com/duanxz/p/3250114.html

作者：威少JAVA
链接：https://www.jianshu.com/p/ed2b56bcba3d
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。