详解ArrayBlockQueue源码解析
今天要讲的是ArrayBlockQueue，ArrayBlockQueue是JUC提供的线程安全的有界的阻塞队列，一看到Array，第一反应：这货肯定和数组有关，既然是数组，那自然是有界的了，我们先来看看ArrayBlockQueue的基本使用方法，然后再看看ArrayBlockQueue的源码。
ArrayBlockQueue基本使用
public static void main(String[] args) throws InterruptedException {
ArrayBlockingQueue<Integer> arrayBlockingQueue=new ArrayBlockingQueue(5);
arrayBlockingQueue.offer(10);
arrayBlockingQueue.offer(50);
arrayBlockingQueue.add(20);
arrayBlockingQueue.add(60);
System.out.println(arrayBlockingQueue);
System.out.println(arrayBlockingQueue.poll());
System.out.println(arrayBlockingQueue);
System.out.println(arrayBlockingQueue.take());
System.out.println(arrayBlockingQueue);
System.out.println(arrayBlockingQueue.peek());
System.out.println(arrayBlockingQueue);
}
运行结果：
- 创建了一个长度为5的ArrayBlockQueue。
- 用offer方法，向ArrayBlockQueue添加了两个元素，分别是10，50。
- 用put方法，向ArrayBlockQueue添加了两个元素，分别是20，60。
- 打印出ArrayBlockQueue，结果是10，50，20，60。
- 用poll方法，弹出ArrayBlockQueue第一个元素，并且打印出来：10。
- 打印出ArrayBlockQueue，结果是50，20，60。
- 用take方法，弹出ArrayBlockQueue第一个元素，并且打印出来：50。
- 打印出ArrayBlockQueue，结果是20，60。
- 用peek方法，弹出ArrayBlockQueue第一个元素，并且打印出来：20。
- 打印出ArrayBlockQueue，结果是20，60。
代码比较简单，但是你肯定会有疑问
- offer/add（在上面的代码中没有演示）/put都是往队列里面添加元素，区别是什么？
- poll/take/peek都是弹出队列的元素，区别是什么？
- 底层代码是如何保证线程安全的？
- 数据保存在哪里？
要解决上面几个疑问，最好的办法当然是看下源码，通过亲自阅读源码所产生的印象远远要比看视频，看博客，死记硬背最后的结论要深刻的多。就算真的忘记了，只要再看看源码，瞬间可以回忆起来。
ArrayBlockQueue源码解析
构造方法
ArrayBlockQueue提供了三个构造方法，如下图所示：
ArrayBlockingQueue(int capacity)
public ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
这是最常用的构造方法，传入capacity，capacity是容量的意思，也就是ArrayBlockingQueue的最大长度，方法内部直接调用了第二个构造方法，传入的第二个参数为false。
ArrayBlockingQueue(int capacity, boolean fair)
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();
}
这个构造方法接受两个参数，分别是capacity和fair，fair是boolean类型的，代表是公平锁，还是非公平锁，可以看出如果我们用第一个构造方法来创建ArrayBlockingQueue的话，采用的是非公平锁，因为公平锁会损失一定的性能，在没有充足的理由的情况下，是没有必要采用公平锁的。
方法内部做了几件事情：
- 创建Object类型的数组，容量为capacity，并且赋值给当前类对象的items。
- 创建排他锁。
- 创建条件变量notEmpty 。
- 创建条件变量notFull。
至于排他锁和两个条件变量是做什么用的，看到后面就明白了。
ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c)
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {
//调用第二个构造方法，方法内部就是初始化数组，排他锁，两个条件变量
this(capacity, fair);
final ReentrantLock lock = this.lock;
lock.lock(); // 开启排他锁
try {
int i = 0;
try {
// 循环传入的集合，把集合中的元素赋值给items数组，其中i会自增
for (E e : c) {
checkNotNull(e);
items[i++] = e;
}
} catch (ArrayIndexOutOfBoundsException ex) {
throw new IllegalArgumentException();
}
count = i;//把i赋值给count
//如果i==capacity，也就是到了最大容量，把0赋值给putIndex，否则把i赋值给putIndex
putIndex = (i == capacity) ? 0 : i;
} finally {
lock.unlock();//释放排他锁
}
}
- 调用第二个构造方法，方法内部就是初始化数组items，排他锁lock，以及两个条件变量。
- 开启排他锁。
- 循环传入的集合，将集合中的元素赋值给items数组，其中i会自增。
- 把i赋值给count。
- 如果i==capacity，说明到了最大的容量，就把0赋值给putIndex，否则把i赋值给putIndex。
- 在finally中释放排他锁。
看到这里，我们应该明白这个构造方法的作用是什么了，就是把传入的集合作为ArrayBlockingQueuede初始化数据，但是我们又会有一个新的疑问：count，putIndex 是做什么用的。
offer(E e)
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();//开启排他锁
try {
if (count == items.length)//如果count==items.length，返回false
return false;
else {
enqueue(e);//入队
return true;//返回true
}
} finally {
lock.unlock();//释放锁
}
}
- 开启排他锁。
- 如果count==items.length，也就是到了最大的容量，返回false。
- 如果count<items.length，执行入队方法，并且返回true。
- 释放排他锁。
看到这里，我们应该可以明白了，ArrayBlockQueue是如何保证线程安全的，还是利用了ReentrantLock排他锁，count就是用来保存数组的当前大小的。我们再来看看enqueue方法。
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
这方法比较简单，在代码里面就不写注释了，做了如下的操作：
- 把x赋值给items[putIndex] 。
- 将putIndex进行自增，如果自增后的值 == items.length，把0赋值给putIndex 。
- 执行count++操作。
- 调用条件变量notEmpty的signal方法，说明在某个地方，必定调用了notEmpty的await方法，这里就是唤醒因为调用notEmpty的await方法而被阻塞的线程。
这里就解答了一个疑问：putIndex是做什么的，就是入队元素的下标。
add(E e)
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");
}
这个方法内部最终还是调用的offer方法。
put(E e)
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();//开启响应中断的排他锁
try {
while (count == items.length)//如果队列满了，调用notFull的await
notFull.await();
enqueue(e);//入队
} finally {
lock.unlock();//释放排他锁
}
}
- 开启响应中断的排他锁，如果在获取锁的过程中，当前的线程被中断，会抛出异常。
- 如果队列满了，调用notFull的await方法，说明在某个地方，必定调用了notFull的signal方法来唤醒当前线程，这里用while循环是为了防止虚假唤醒。
- 执行入队操作。
- 释放排他锁。
可以看到put方法和 offer/add方法的区别了：
- offer/add：如果队列满了，直接返回false。
- put：如果队列满了，当前线程被阻塞，等待唤醒。
poll()
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
- 开启排他锁。
- 如果count==0，直接返回null，否则执行dequeue出队操作。
- 释放排他锁。
我们来看dequeue方法：
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];//获得元素的值
items[takeIndex] = null;//把null赋值给items[takeIndex]
if (++takeIndex == items.length)//如果takeIndex自增后的值== items.length，就把0赋值给takeIndex
takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();//唤醒因为调用notFull的await方法而被阻塞的线程
return x;
}
- 获取元素的值，takeIndex保存的是出队的下标。
- 把null赋值给items[takeIndex]，也就是清空被弹出的元素。
- 如果takeIndex自增后的值== items.length，就把0赋值给takeIndex。
- count--。
- 唤醒因为调用notFull的await方法而被阻塞的线程。
这里调用了notFull的signal方法来唤醒因为调用notFull的await方法而被阻塞的线程，那到底在哪里调用了notFull的await方法呢，还记不记得在put方法中调用了notFull的await方法，我们再看看：
while (count == items.length)
notFull.await();
当队列满了，就调用 notFull.await()来等待，在出队操作中，又调用了notFull.signal()来唤醒。
take()
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
- 开启排他锁。
- 如果count==0，代表队列是空的，则调用notEmpty的await方法，用while循环是为了防止虚假唤醒。
- 执行出队操作。
- 释放排他锁。
这里调用了notEmpty的await方法，那么哪里调用了notEmpty的signal方法呢？在enqueue入队方法里。
我们可以看到take和poll的区别：
- take：如果队列为空，会阻塞，直到被唤醒了。
- poll： 如果队列为空，直接返回null。
peek()
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex);
} finally {
lock.unlock();
}
}
final E itemAt(int i) {
return (E) items[i];
}
- 开启排他锁。
- 获得元素。
- 释放排他锁。
我们可以看到peek和poll/take的区别：
- peek，只是获取元素，不会清空元素。
- poll/take，获取并清空元素。
size()
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
- 开启排他锁。
- 返回count。
- 释放排他锁。
总结
至此，ArrayBlockQueue的核心源码就分析完毕了，我们来做一个总结：
- ArrayBlockQueue有几个比较重要的字段，分别是items，保存的是队列的数据，putIndex保存的是入队的下标，takeIndex保存的是出队的下标，count用来统计队列元素的个数，lock用来保证线程的安全性，notEmpty和notFull两个条件变量实现唤醒和阻塞。
- offer和add是一样的，其中add方法内部调用的就是offer方法，如果队列满了，直接返回false。
- put，如果队列满了，会被阻塞。
- peek，只是弹出元素，不会清空元素。
- poll，弹出并清空元素，如果队列为空，直接返回null。
- take，弹出并清空元素，如果队列为空，会被阻塞。
以上所述是小编给大家介绍的ArrayBlockQueue源码解析详解整合，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对脚本之家网站的支持！
相关文章
JDK1.7 之java.nio.file.Files 读取文件仅需一行代码实现
下面小编就为大家分享一篇JDK1.7 之java.nio.file.Files 读取文件仅需一行代码实现，具有很好的参考价值，希望对大家有所帮助2017-11-11
本篇文章主要介绍了Sprigmvc项目转为springboot的方法，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2018-02-02
解决spring 处理request.getInputStream()输入流只能读取一次问题
这篇文章主要介绍了解决spring 处理request.getInputStream()输入流只能读取一次问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-09-09
这篇文章主要介绍了浅谈Spring中Bean的作用域,生命周期和注解，从创建到消亡的完整过程,例如人从出生到死亡的整个过程就是一个生命周期。本文将通过示例为大家详细讲讲，感兴趣的可以学习一下2022-06-06
这篇文章主要为大家详细介绍了Java实现解析.xlsb文件的相关方法，文中的示例代码讲解详细，具有一定的借鉴价值，感兴趣的可以了解一下2023-01-01
AQS内部维护着一个FIFO队列，该队列就是CLH同步队列。下面小编来简单介绍下这个队列2019-05-05
这篇文章主要介绍了java之向linux文件夹下写文件无权限的问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-09-09
Redis 除了做缓存，还能干很多很多事情：分布式锁、限流、处理请求接口幂等性。。。太多太多了～今天想和小伙伴们聊聊用 Redis 处理接口限流，这也是最近的 项目涉及到这个知识点了，我就拎出来和大家聊聊这个话题2022-07-07
这篇文章主要给大家介绍了关于java中最易犯错的特殊字符的相关资料，文中通过示例代码介绍的非常详细，对大家学习或者使用java具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧2019-09-09
这篇文章主要介绍了如何让Spring Rest 接口中路径参数可选，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2021-10-10
最新评论