结构
ArrayBlockingQueue 的内部有一个数组 items ,用来存放队列元素,
putindex 变量表示入队元素下标, takelndex 是出队下标, cou nt 统计队列元素个数 。 从定
义可知,这些变量并没有使用 volatile 修饰,这是因为访问这些变量都是在锁块内,而加
锁己经保证了锁块内变量的内存可见性了 。 另外有个独占锁 lock 用来保证出、入队操作的
原子性,这保证了同时只有一个线程可以进行入队、出队操作 。 另外, notEmpty 、 notFull
条件变量用来进行出、入队的同步。
由于 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();
}
在默认情况下使用 ReentrantLock 提供的非公平独占锁进行出、入队操作的同步。
offer 操作
向队列尾部插入一个元素, 如果队列有空 闲 空 间则插入成功后返回 true,如果队列己
满则丢弃当前元素然后返回 false 。 如果 e 元素为 null 则抛出 NullPointerException 异常 。另 外,
该方法是不阻塞的 。
public boolean offer(E e) {
//1 检查元素是否为空,为空就抛出空指针
checkNotNull(e);
//2 获取独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//3 如果队列满则返回 false
if (count == items.length)
return false;
else {
//4 否则插入元素
enqueue(e);
return true;
}
} finally {
//5
lock.unlock();
}
}
代码- 2 获取独占锁 , 当前线程获取该锁后 , 其他入队和出 队操作的线程都会被阻
塞挂起而后被放入 lock 锁的 AQS 阻塞队列 。
代码-3 判 断如果队列满则直接返回 false,否则调用 enqueue 方法后返回 true,
enqueue 的代码如下。
- 入队
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
// 6 元素入队
final Object[] items = this.items;
items[putIndex] = x;
// 7 计算下一个元素应该存放的下标位置
if (++putIndex == items.length)
putIndex = 0;
count++;
// 8
notEmpty.signal();
}
首先把当前元素放入 items 数组 , 然后计算下一个元素应该存放的下标位置 ,
并递增元素个数计数器,最后激活 notEmpty 的条件 队列 中 因为调用 take 操作而被阻塞 的
一个线程。这里由 于在操作共享变量 count 前加 了锁,所以不存在内存不可见问题,加过
锁后获取的共享变量都是从主内存获取的 , 而不是从 CPU 缓存或者寄存器获取。
代码5 释放锁 , 然后会把修改的共享变量值( 比如 count 的值〉刷新回主内存中,
这样其他线程通过加锁再次读取这些共享变量时,就可以看到最新的值。
put 操作-可中断
向队列尾部插入一个元素 ,如果 队列有空 闲则插入后直接返 回 true,如果 队列 己满则
阻塞当前线程直到队列有空闲井插入成功后返回 true,如果在阻塞时被其他线程设置了 中
断标志, 则被阻塞线程会抛出 InteηuptedException 异常而返回。另 外, 如果 e 元素为 null
则抛出 NullPointerException 异常 。
public void put(E e) throws InterruptedException {
//1. 验证是否为空
checkNotNull(e);
//2.获取独占锁 可中断
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//3 如查队列满,则把当前线程放入notFull管理的条件队列
while (count == items.length)
notFull.await();
//4 入队
enqueue(e);
} finally {
//5 解锁
lock.unlock();
}
}
在代码 ( 2 )中,在获取锁的过程 中 当 前线程被其他线程中 断 了 ,则 当 前线程会抛 出
InterruptedException 异常而退出 。
代码 ( 3 ) 判断如果 当前队列己满 , 则把当前线程阻塞挂起后放入 notFull 的条件队列 ,
注意这里也是使用了 whi le 循环而不是 if 语句 。
代码( 4 ) 判断如果队列不满则插入当前元素
poll 操作-不可中断
从队列头部获取并移除一个元素 ,如果队列为空则返回 null , 该方法是不阻塞的
public E poll() {
//1. 获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//2 当前队列为空则返回 null , 否则调用 dequeue获取
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
- 出队
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
// 4获取元素值
E x = (E) items[takeIndex];
//5 数组中的值为 null
items[takeIndex] = null;
//6 队头指针再次计算,队列元素个数减 1
if (++takeIndex == items.length)
takeIndex = 0;
//队列元素个数减 1
count--;
if (itrs != null)
itrs.elementDequeued();
//发送信号激活notFull条件队列 里面的一个线程
notFull.signal();
return x;
}
首先获取当前队头元素并将其保存到局部变量 ,然后重置队头元素
为 null,并重新设置队头下标,递减元素计数器, 最后发送信号激活 notFull 的条件队列
里面一个因为调用 put 方法而被阻塞的线程
take 操作-可中断
public E take() throws InterruptedException {
//1.获取锁
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
//2 队列为空,则等待,直到队列中有元素
while (count == 0)
notEmpty.await();
//3 出队列 获取队头元素
return dequeue();
} finally {
//4 解锁
lock.unlock();
}
}
如果队列为
空则把当前线程挂起后放入 notEmpty 的条件队列 , 等其他线程调用 notEmpty.signal() 方法
后再返回。需要注意的是,这里也是使用 while 循环进行检测井等待而不是使用 if 语句。
peek操作
获取队列头部元素但是不从队列里面移除它 , 如果队列为空则返回 null , 该方法是不
阻塞的。
public E peek() {
//1. 获取锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//2. 获取头元素
return itemAt(takeIndex); // null when queue is empty
} finally {
//3 .解锁
lock.unlock();
}
}
- itemAt方法
final E itemAt(int i) {
return (E) items[i];
}
首先获取独 占锁 , 然后 从数组 items 中 获取 当 前队头 下标 的值并
返回 , 在返回前释放获取的锁
size 操作
public int size() {
//1 获取独占锁
final ReentrantLock lock = this.lock;
lock.lock();
try {
//2. 返回元素个数
return count;
} finally {
//3 解锁
lock.unlock();
}
}
获取锁后直接返回 count ,并在返回前释放锁 。 也许你会问,这
里又没有修改 count 的 值,只是简单地获取,为何要加锁呢?其实如果 count 被声明
为 volatile 的这里就不需要加锁了,因为 volatile 类型的变量保证了 内存的可见性,而
ArrayBlockingQueue 中的 count 并没有被 声 明为 volatile 的,这是因为 count 操作都是在获
取锁后进行的,而获取锁的语义之一是,获取锁后访问的变量都是从主内存获取的,这保
证了变量的内存可见性 。
ArrayBlockingQueue 通过使用全局独占锁实现了同时只能有一个线
程进行入队或者出队操作,这个锁的粒度比较大,有点类似于在方法上添加 synchronized
的意思 。 其中 。他r 和 poll 操作通过简单的加锁进行入队、出队操作,而 put 、 take 操
作则使用条件变量实现了,如果队列满则等待,如果队列空则等待,然后分别在出
队和入队操作中发送信号激活等待线程实现同步。另 外,相比 LinkedBlockingQueue,
ArrayBlockingQueue 的 size 操作的结果是精确的 , 因为计算前加了全局锁。