并发集合类(4):ArrayBlockingQueue

什么是ArrayBlockingQueue？

ArrayBlockingQueue是JUC里的一个有界阻塞队列，底层基于数组实现，非常适合用来做生产者-消费者模型。

ArrayBlockingQueue的设计和用途和LinkedBlockingQueue类似，只是底层实现不同。

相同之处:

1. 都实现了Blocking接口，支持阻塞的put和take方法
2. 内部使用ReentrantLock来实现线程安全
3. 都可以设置容量，作为有界队列

不同之处：

1. 底层一个是数组实现，一个是链表实现
2. LinkedBlockingQueue可以作为无界队列来使用
3. ArrayBlockingQueue内部只有一把锁，读写互斥，因此并发度更低，而LinkedBlockingQueue内部有两把锁，读写并行，并发度更高。
4. ArrayBlockingQueue可以实现公平队列

ArrayBlockingQueue的内部数据结构

final Object[] items; ## int takeIndex; int putIndex; int count; final ReentrantLock lock; private final Condition notEmpty; private final Condition notFull; transient Itrs itrs;

items是真正存放队列元素的数组，逻辑上是环形数组。

takeIndex和putIndex是位置指针，一个指向队头，一个指向队尾。count则是用来判空和满的。

lock用于内部并发控制，notEmpty和notFull是对应的条件队列。

itrs管理所有活跃迭代器的共享状态。

ArrayBlockingQueue的方法

构造方法

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

在ArrayBlockingQueue的构造方法中，可以设置队列的大小以及是否为公平队列。

offer操作

效果同LinkedBlockingQueue的offer操作。

public boolean offer(E e) { Objects.requireNonNull(e); final ReentrantLock lock = this.lock; lock.lock(); try { if (count == items.length) return false; else { enqueue(e); return true; } } finally { lock.unlock(); } } private void enqueue(E e) { // assert lock.isHeldByCurrentThread(); // assert lock.getHoldCount() == 1; // assert items[putIndex] == null; final Object[] items = this.items; items[putIndex] = e; if (++putIndex == items.length) putIndex = 0; count++; notEmpty.signal(); }

1. 加锁
2. 如果队列已满，则直接返回false
3. 加入队列，唤醒阻塞的take线程
4. 解锁

put操作

put操作也与LinkedBlockingQueue类似

public void put(E e) throws InterruptedException { Objects.requireNonNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); } }

1. 加锁
2. 当队列已满的时候，放入条件队列中
3. 入队
4. 解锁

poll操作

public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } }

1. 加锁
2. 如果队列为空，返回空值，否则出队
3. 解锁

take操作

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

1. 加锁
2. 如果队列为空，则进入条件队列
3. 出队
4. 解锁

size方法

public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return count; } finally { lock.unlock(); } }

ArrayBlockingQueue的size()方法是加锁的，因为count并没有被volatile所修饰，需要使用加锁的语义来获取最新的count。