- 👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
- 📕系列专栏：Java设计模式、数据结构和算法、Kafka从入门到成神、Kafka从成神到升仙、Spring从成神到升仙系列
- 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
- 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
- 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀
文章目录
- 从源码全面解析 ArrayBlockingQueue 的来龙去脉
- 一、引言
- 二、使用
- 三、源码
- 1、初始化
- 2、生产者的源码
- 2.1 add()源码实现
- 2.2 offer()源码实现
- 2.3 offer(time,unit)源码实现
- 2.4 put()源码实现
- 2.5 final ReentrantLock lock = this.lock
- 2.6 虚假唤醒
- 3、消费者的源码
- 3.1 remove()源码实现
- 3.2 poll() 源码实现
- 3.3 poll(time,unit)源码实现
- 3.4 take()源码实现
- 四、流程图
- 五、总结
从源码全面解析 ArrayBlockingQueue 的来龙去脉
一、引言
并发编程在互联网技术使用如此广泛，几乎所有的后端技术面试官都要在并发编程的使用和原理方面对小伙伴们进行 360° 的刁难。
作为一个在互联网公司面一次拿一次 Offer 的面霸，打败了无数竞争对手，每次都只能看到无数落寞的身影失望的离开，略感愧疚（请允许我使用一下夸张的修辞手法）。
于是在一个寂寞难耐的夜晚，暖男我痛定思痛，决定开始写 《吊打面试官》 系列，希望能帮助各位读者以后面试势如破竹，对面试官进行 360° 的反击，吊打问你的面试官，让一同面试的同僚瞠目结舌，疯狂收割大厂 Offer！
虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马
二、使用
对于阻塞队列，想必大家应该都不陌生，我们这里简单的介绍一下，对于 Java 里面的阻塞队列，其使用了 **生产者和消费者 **的模型
对于生产者来说，主要有以下几部分：
add(E) // 添加数据到队列，如果队列满了，无法存储，抛出异常
offer(E) // 添加数据到队列，如果队列满了，返回false
offer(E,timeout,unit) // 添加数据到队列，如果队列满了，阻塞timeout时间，如果阻塞一段时间，依然没添加进入，返回false
put(E) // 添加数据到队列，如果队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等！
对于消费者来说，主要有以下几部分：
remove() // 从队列中移除数据，如果队列为空，抛出异常
poll() // 从队列中移除数据，如果队列为空，返回null，么的数据
poll(timeout,unit) // 从队列中移除数据，如果队列为空，挂起线程timeout时间，等生产者扔数据，再获取
take() // 从队列中移除数据，如果队列为空，线程挂起，一直等到生产者扔数据，再获取
我们本篇来讲讲堵塞队列中的第一员猛将，ArrayBlockingQueue 的故事
我们简单来写一个小demo
public class ArrayBlockingQueueTest {public static void main(String[] args) throws Exception {// 必须设置队列的长度ArrayBlockingQueue queue = new ArrayBlockingQueue(4);// 生产者扔数据queue.add("1");queue.offer("2");queue.offer("3", 2, TimeUnit.SECONDS);queue.put("2");// 消费者取数据System.out.println(queue.remove());System.out.println(queue.poll());System.out.println(queue.poll(2, TimeUnit.SECONDS));System.out.println(queue.take());}
}
三、源码
1、初始化
由于我们的 ArrayBlockingQueue 底层使用的是数据结构，所以我们需要在初始化的时候指定其大小，如下：
// 设置其大小长度为 4
ArrayBlockingQueue queue = new ArrayBlockingQueue(4);// 初始化
public ArrayBlockingQueue(int capacity) {this(capacity, false);
}// 初始化ArrayBlockingQueue的一些初始变量
public ArrayBlockingQueue(int capacity, boolean fair) {// 如果传一个负数，直接完蛋if (capacity <= 0)throw new IllegalArgumentException();// 初始化数组itemsthis.items = new Object[capacity];// 初始化lock非公平锁lock = new ReentrantLock(fair);// 消费者挂起线程和唤醒线程用到的ConditionnotEmpty = lock.newCondition();// 生产者挂起线程和唤醒线程用到的ConditionnotFull = lock.newCondition();
}
除了我们初始化的这些变量，也有其他的一些变量：
// 存储数据的下标
int putIndex
// 取数据的下标
int takeIndex
// 当前数组中存储的数据长度
int count
对于 ReentrantLock 和 newCondition 的知识点，可以看以下博文：
- newCondition的源码分析
- ReentrantLock的源码分析
2、生产者的源码
2.1 add()源码实现
public boolean add(E e) {return super.add(e);
}// 走到这里会发现，我们的add方法就是调用了offer方法
// offer: 添加数据到队列，如果队列满了，返回false
// 所以这里offer满了，就会抛出异常："Queue full"
public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full");
}
2.2 offer()源码实现
public boolean offer(E e) {// 检测当前的入参是否为null// 如果是的话直接抛出异常checkNotNull(e);//【面试绝杀招】为什么会这样引用使用？final ReentrantLock lock = this.lock;// 直接加锁，保证线程安全lock.lock();try {// 如果当前数组存储的长度等于总容量// 直接返回false,插入失败if (count == items.length)return false;else {// 插入enqueue(e);return true;}} finally {// 结束之后将锁释放掉lock.unlock();}
}// 添加数据
private void enqueue(E x) {// 我们发现，这引用又来了final Object[] items = this.items;// 当前数组的赋值items[putIndex] = x;// 如果下标等于我们的总容量，需要重新置下标值为0if (++putIndex == items.length)putIndex = 0;// 数组容量加一count++;// 唤醒消费者等待的线程notEmpty.signal();
}
2.3 offer(time,unit)源码实现
生产者在添加数据时，如果队列已经满了，阻塞一会
- 阻塞到消费者消费了消息，然后唤醒当前阻塞线程
- 阻塞到了
time时间，再次判断是否可以添加，不能，直接告辞
public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {// 检测当前的入参是否为空checkNotNull(e);// 统一时间格式long nanos = unit.toNanos(timeout);// 持有引用final ReentrantLock lock = this.lock;// 允许的中断的加锁方式lock.lockInterruptibly();try {// 如果当前的存储等于数组的长度// 这里为什么不能用if判断,需要用while,牵扯到虚假唤醒,我们后面聊while (count == items.length) {// 时间小于0,直接返回falseif (nanos <= 0)return false;// 挂起当前线程nanos = notFull.awaitNanos(nanos);}// 添加到数组中enqueue(e);return true;} finally {// 解锁lock.unlock();}
}
2.4 put()源码实现
public void put(E e) throws InterruptedException {// 检测当前的入参是否为空checkNotNull(e);// 持有引用final ReentrantLock lock = this.lock;// 允许的中断的加锁方式lock.lockInterruptibly();try {// 如果当前的存储等于数组的长度// 这里为什么不能用if判断,需要用while,牵扯到虚假唤醒,我们后面聊while (count == items.length)// 无时间挂起当前线程notFull.await();// 添加到队列enqueue(e);} finally {// 解锁lock.unlock();}
}
通过上面的源码分析，我们应该可以理解上面说的这几句话了
add(E) // 添加数据到队列，如果队列满了，无法存储，抛出异常
offer(E) // 添加数据到队列，如果队列满了，返回false
offer(E,timeout,unit) // 添加数据到队列，如果队列满了，阻塞timeout时间，如果阻塞一段时间，依然没添加进入，返回false
put(E) // 添加数据到队列，如果队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等！
接着我们讲两个小细节，也是面试震惊面试官的地方
2.5 final ReentrantLock lock = this.lock
在我们 Doug Lea 里写的代码中，java.util.concurrent 包下 和 HashMap 中都有类似的写法
这种写法到底有什么好处呢，为什么我们不能直接使用成员变量 lock 来进行加锁解锁
感兴趣的可以看下这篇博文：原因
不感兴趣的可以看我的总结分析：
首先我们需要准备下面两个代码，将其反编译得到 Java 字节码
-
引用状态
public void get1(){final ReentrantLock lock = this.lock;lock.lock(); } -
非引用状态
public void get2(){lock.lock(); }
通过对比字节码我们发现，引用状态的字节码相较于非引用状态少了一个指令：getfield
而这个缺少的指令，也正是 Doug Lea 优化的来源：从栈读变量比从堆读变量会更cache-friendly，本地变量最终绑定到CPU寄存器的可能性更高。
但由于现在的 Java 编译器已经非常先进了，不论采用哪种方式，最终形成的机器指令都是一样的
所以，Doug Lea 的优化在之前是字节码层面的优化，但如今确实没有卵用
2.6 虚假唤醒
我们上面有一段 while 循环的代码：
// 如果当前的存储等于数组的长度
// 这里为什么不能用if判断,需要用while,牵扯到虚假唤醒,我们后面聊
while (count == items.length) {// 无时间挂起当前线程notFull.await();
}
// 添加到队列
enqueue(e);
我们 A 线程判断数组内还有空余，则放入数组
我们 B 线程判断其 count == items.length 进入挂起状态，当我们的 B 线程被唤醒时，如果不经历 count == items.length 的过程，就会将我们 A 线程的 3 数据给覆盖掉
3、消费者的源码
3.1 remove()源码实现
- 主要使用了我们的poll方法
public E remove() {// 直接调用poll方法E x = poll();// 如果有数据则返回// 无数据则抛出异常if (x != null)return x;elsethrow new NoSuchElementException();
}
3.2 poll() 源码实现
public E poll() {// 还是引用final ReentrantLock lock = this.lock;// 锁一下lock.lock();try {// 判断数组容量是否等于0(数组无容量),返回null// 如果数组中有数据,则进行dequeue方法return (count == 0) ? null : dequeue();} finally {// 解锁lock.unlock();}
}private E dequeue() {// 还是引用final Object[] items = this.items;// 当前弹出数组的下标E x = (E) items[takeIndex];// 弹出后将当前下标的数据置为空items[takeIndex] = null;// 如果我们的弹出下标和我们数组的大小一样时,需要更新弹出下标if (++takeIndex == items.length)takeIndex = 0;// 数组数据数量减一count--;// 迭代器内容，先忽略if (itrs != null)itrs.elementDequeued();// 唤醒生产者的线程notFull.signal();return x;
}
3.3 poll(time,unit)源码实现
public E poll(long timeout, TimeUnit unit) throws InterruptedException {// 将时间转化成统一单位long nanos = unit.toNanos(timeout);// 引用final ReentrantLock lock = this.lock;// 可中断的加锁lock.lockInterruptibly();try {// 看一下当前的数组还有容量没while (count == 0) {// 如果没有容量并且时间也到期了，返回nullif (nanos <= 0)return null;// 进入带有时间的等待状态(扔到Condition队列中)nanos = notEmpty.awaitNanos(nanos);}// 被唤醒后并且当前的数组有容量// 弹出队列中的数据即可return dequeue();} finally {// 解锁lock.unlock();}
}
3.4 take()源码实现
public E take() throws InterruptedException {// 引用final ReentrantLock lock = this.lock;// 可中断的加锁lock.lockInterruptibly();try {// 看一下当前的数组还有容量没while (count == 0){// 没容量直接扔Condition队列等待notEmpty.await();}// 被唤醒后并且当前的数组有容量// 弹出队列中的数据即可return dequeue();} finally {// 解锁lock.unlock();}
}
四、流程图
私聊我获取高清流程图
五、总结
鲁迅先生曾说：独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。
其实很多时候，并不是我们不够努力，很可能就是自己努力的方向不对，如果有一个人能稍微指点你一下，你真的可能会少走几年弯路。
如果你也对 后端架构和中间件源码 有兴趣，欢迎添加博主微信：hls1793929520，一起学习，一起成长
我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，喜欢后端架构和中间件源码。
我们下期再见。
我从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。
往期文章推荐：
- 从根上剖析ReentrantLock的来龙去脉
- 阅读完synchronized和ReentrantLock的源码后，我竟发现其完全相似
- 从源码全面解析 ThreadLocal 关键字的来龙去脉
- 从源码全面解析 synchronized 关键字的来龙去脉
- 阿里面试官让我讲讲volatile，我直接从HotSpot开始讲起，一套组合拳拿下面试
