ArrayBlockingQueue顾名思义,是基于数组实现的阻塞队列,阻塞队列可以用于缓冲任务,实现生产者消费者模式,例如线程池中的工作队列。那么怎么才能用数组实现阻塞队列呢?
我们先从ArrayBlockingQueue的功能说起
我们先拆解一下问题,把这个问题分成
如何用数组实现数据的入队出队操作呢,很多同学第一时间会想到,通过一个index字段存储当前数组下一个写入的位置。那么如何处理出队呢,很多同学会想到出队就是返回数组第一个元素,并且把后面所有的元素向前移动一位。
这个方案的问题在于出队时会移动大量的元素,它的时间复杂度是O(n),那有没有更高效的方案呢?还有另一个循环数组的方案,我们通过两个int字段,分别记录下一个要入队和下一个要出队的元素的位置,当入队到数组末尾时,从0开始,同样当出队到末尾时,也从0开始。
另外当队列为空和队列已满的时候,takeIndex和putIndex都指向相同的位置,所以为了进行区分,我们可以用一个count字段存储队列元素数量,这样当count=0的时候说明队列为0,count=数组容量的时候说明队列已满
下面代码展示了用数组实现队列的具体实现。
class ArrayBlockingQueue<E> { final Object[] items; int takeIndex; int putIndex; int count; public ArrayBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; } private void enqueue(E e) { Object[] items = this.items; items[putIndex] = e; if (++putIndex == items.length) putIndex = 0; count++; } private E dequeue() { Object[] items = this.items; E e = (E) items[takeIndex]; items[takeIndex] = null; if (++takeIndex == items.length) takeIndex = 0; count--; return e; } }
「在队列已满时,对入队的请求进行阻塞,当队列有剩余空间时,释放入队请求」这个需求本质上是一个条件等待的特例,写入的条件是队列不满,不满足条件的时候需要等待,直到满足条件为止。
在Java中,实现条件等待有synchronized+Object.wait和Lock+Condition.await两种方式,这里不用synchronized方案,是因为
通过Lock和Condition的方案,还能够保证线程安全,因为上面的环形数组实现中,线程间共享的变量有items数组、takeIndex、putIndex、count,线程安全涉及到原子性可见性重排序几个方面,通过Lock类加锁可以对共享变量的读写操作进行保护。
定义阻塞的Lock对象和Condition,条件分为不满和不空两个条件。
class ArrayBlockingQueue<E> { final Object[] items; int takeIndex; int putIndex; int count; ReentrantLock lock; private final Condition notEmpty; private final Condition notFull; public ArrayBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; // 创建lock对象 lock = new ReentrantLock(); // 创建非空的Condition notEmpty = lock.newCondition(); // 创建不满的Condition notFull = lock.newCondition(); } }
以入队操作添加实现为例,能够入队的条件是队列不满,也就是count < items.length,不能入队的条件反过来就是count == items.length。 当满足条件后,我们就可以入队了,入队之后,还需要唤醒等待出队的线程。
put方法的流程为
public void put(E e) throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); enqueue(e); notEmpty.signal(); } finally { lock.unlock(); } }
方法中还有一些小细节
ReentrantLock lock = this.lock;
这是因为如果不使用局部变量,后面所有使用实例变量的调用,在字节码指令层面需要变成先调用aload 0获取到this,再调用getField指令获取字段值,再进行其他操作。而先把lock存到局部变量中,后面所有的获取lock就可以变成一个aload xxx指令,从而节省了指令数量,也就会加快方法的执行速度。
如果不放在锁内,则可能会出现多个线程同时看到满足条件,进而去加锁入队。虽然入队还是在临界区,但是会出现队列已满,仍然在执行入队操作的情况。这个问题和单例的double check locking中少些一个check的问题类似。
take方法是和put相对应的出队方法,和put流程基本一致
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); E element = dequeue(); notFull.signal() return element; } finally { lock.unlock(); } }