从源码全面解析 ArrayBlockingQueue 的来龙去脉

• 👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
• 📕系列专栏：Java设计模式、数据结构和算法、Kafka从入门到成神、Kafka从成神到升仙、Spring从成神到升仙系列
• 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
• 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
• 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

文章目录

从源码全面解析 ArrayBlockingQueue 的来龙去脉

一、引言

并发编程在互联网技术使用如此广泛，几乎所有的后端技术面试官都要在并发编程的使用和原理方面对小伙伴们进行 360° 的刁难。

作为一个在互联网公司面一次拿一次

Offer

的面霸，打败了无数竞争对手，每次都只能看到无数落寞的身影失望的离开，略感愧疚（请允许我使用一下夸张的修辞手法）。

于是在一个寂寞难耐的夜晚，暖男我痛定思痛，决定开始写 《吊打面试官》 系列，希望能帮助各位读者以后面试势如破竹，对面试官进行 360° 的反击，吊打问你的面试官，让一同面试的同僚瞠目结舌，疯狂收割大厂

Offer

！

虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马

二、使用

对于阻塞队列，想必大家应该都不陌生，我们这里简单的介绍一下，对于

Java

里面的阻塞队列，其使用了 **生产者和消费者 **的模型

对于生产者来说，主要有以下几部分：

add(E)// 添加数据到队列，如果队列满了，无法存储，抛出异常offer(E)// 添加数据到队列，如果队列满了，返回falseoffer(E,timeout,unit)// 添加数据到队列，如果队列满了，阻塞timeout时间，如果阻塞一段时间，依然没添加进入，返回falseput(E)// 添加数据到队列，如果队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等！

对于消费者来说，主要有以下几部分：

remove()// 从队列中移除数据，如果队列为空，抛出异常poll()// 从队列中移除数据，如果队列为空，返回null，么的数据poll(timeout,unit)// 从队列中移除数据，如果队列为空，挂起线程timeout时间，等生产者扔数据，再获取take()// 从队列中移除数据，如果队列为空，线程挂起，一直等到生产者扔数据，再获取

我们本篇来讲讲堵塞队列中的第一员猛将，

ArrayBlockingQueue

的故事

我们简单来写一个小

demo

publicclassArrayBlockingQueueTest{publicstaticvoidmain(String[] args)throwsException{// 必须设置队列的长度ArrayBlockingQueue queue =newArrayBlockingQueue(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

底层使用的是数据结构，所以我们需要在初始化的时候指定其大小，如下：

// 设置其大小长度为 4ArrayBlockingQueue queue =newArrayBlockingQueue(4);// 初始化publicArrayBlockingQueue(int capacity){this(capacity,false);}// 初始化ArrayBlockingQueue的一些初始变量publicArrayBlockingQueue(int capacity,boolean fair){// 如果传一个负数，直接完蛋if(capacity <=0)thrownewIllegalArgumentException();// 初始化数组itemsthis.items =newObject[capacity];// 初始化lock非公平锁
    lock =newReentrantLock(fair);// 消费者挂起线程和唤醒线程用到的Condition
    notEmpty = lock.newCondition();// 生产者挂起线程和唤醒线程用到的Condition
    notFull =  lock.newCondition();}

除了我们初始化的这些变量，也有其他的一些变量：

// 存储数据的下标int putIndex
// 取数据的下标int takeIndex
// 当前数组中存储的数据长度int count

对于

ReentrantLock

和

newCondition

的知识点，可以看以下博文：

• newCondition的源码分析
• ReentrantLock的源码分析

2、生产者的源码

2.1 add()源码实现

publicbooleanadd(E e){returnsuper.add(e);}// 走到这里会发现，我们的add方法就是调用了offer方法// offer: 添加数据到队列，如果队列满了，返回false// 所以这里offer满了，就会抛出异常："Queue full"publicbooleanadd(E e){if(offer(e))returntrue;elsethrownewIllegalStateException("Queue full");}

2.2 offer()源码实现

publicbooleanoffer(E e){// 检测当前的入参是否为null// 如果是的话直接抛出异常checkNotNull(e);//【面试绝杀招】为什么会这样引用使用？finalReentrantLock lock =this.lock;// 直接加锁，保证线程安全
    lock.lock();try{// 如果当前数组存储的长度等于总容量// 直接返回false,插入失败if(count == items.length)returnfalse;else{// 插入enqueue(e);returntrue;}}finally{// 结束之后将锁释放掉
        lock.unlock();}}// 添加数据privatevoidenqueue(E x){// 我们发现，这引用又来了finalObject[] items =this.items;// 当前数组的赋值
    items[putIndex]= x;// 如果下标等于我们的总容量，需要重新置下标值为0if(++putIndex == items.length)
        putIndex =0;// 数组容量加一
    count++;// 唤醒消费者等待的线程
    notEmpty.signal();}

2.3 offer(time,unit)源码实现

生产者在添加数据时，如果队列已经满了，阻塞一会

• 阻塞到消费者消费了消息，然后唤醒当前阻塞线程
• 阻塞到了 time 时间，再次判断是否可以添加，不能，直接告辞

publicbooleanoffer(E e,long timeout,TimeUnit unit)throwsInterruptedException{// 检测当前的入参是否为空checkNotNull(e);// 统一时间格式long nanos = unit.toNanos(timeout);// 持有引用finalReentrantLock lock =this.lock;// 允许的中断的加锁方式
    lock.lockInterruptibly();try{// 如果当前的存储等于数组的长度// 这里为什么不能用if判断,需要用while,牵扯到虚假唤醒,我们后面聊while(count == items.length){// 时间小于0,直接返回falseif(nanos <=0)returnfalse;// 挂起当前线程
            nanos = notFull.awaitNanos(nanos);}// 添加到数组中enqueue(e);returntrue;}finally{// 解锁
        lock.unlock();}}

2.4 put()源码实现

publicvoidput(E e)throwsInterruptedException{// 检测当前的入参是否为空checkNotNull(e);// 持有引用finalReentrantLock lock =this.lock;// 允许的中断的加锁方式
    lock.lockInterruptibly();try{// 如果当前的存储等于数组的长度// 这里为什么不能用if判断,需要用while,牵扯到虚假唤醒,我们后面聊while(count == items.length)// 无时间挂起当前线程
            notFull.await();// 添加到队列enqueue(e);}finally{// 解锁
        lock.unlock();}}

通过上面的源码分析，我们应该可以理解上面说的这几句话了

add(E)// 添加数据到队列，如果队列满了，无法存储，抛出异常offer(E)// 添加数据到队列，如果队列满了，返回falseoffer(E,timeout,unit)// 添加数据到队列，如果队列满了，阻塞timeout时间，如果阻塞一段时间，依然没添加进入，返回falseput(E)// 添加数据到队列，如果队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等！

接着我们讲两个小细节，也是面试震惊面试官的地方

2.5 final ReentrantLock lock = this.lock

在我们

Doug Lea

里写的代码中，

java.util.concurrent

包下 和

HashMap

中都有类似的写法

这种写法到底有什么好处呢，为什么我们不能直接使用成员变量

lock

来进行加锁解锁

感兴趣的可以看下这篇博文：原因

不感兴趣的可以看我的总结分析：

首先我们需要准备下面两个代码，将其反编译得到

Java

字节码

• 引用状态publicvoidget1(){finalReentrantLock lock =this.lock; lock.lock();}
• 非引用状态publicvoidget2(){ 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方法

publicEremove(){// 直接调用poll方法E x =poll();// 如果有数据则返回// 无数据则抛出异常if(x !=null)return x;elsethrownewNoSuchElementException();}

3.2 poll() 源码实现

publicEpoll(){// 还是引用finalReentrantLock lock =this.lock;// 锁一下
    lock.lock();try{// 判断数组容量是否等于0(数组无容量),返回null// 如果数组中有数据,则进行dequeue方法return(count ==0)?null:dequeue();}finally{// 解锁
        lock.unlock();}}privateEdequeue(){// 还是引用finalObject[] 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)源码实现

publicEpoll(long timeout,TimeUnit unit)throwsInterruptedException{// 将时间转化成统一单位long nanos = unit.toNanos(timeout);// 引用finalReentrantLock lock =this.lock;// 可中断的加锁
    lock.lockInterruptibly();try{// 看一下当前的数组还有容量没while(count ==0){// 如果没有容量并且时间也到期了，返回nullif(nanos <=0)returnnull;// 进入带有时间的等待状态(扔到Condition队列中)
            nanos = notEmpty.awaitNanos(nanos);}// 被唤醒后并且当前的数组有容量// 弹出队列中的数据即可returndequeue();}finally{// 解锁
        lock.unlock();}}

3.4 take()源码实现

publicEtake()throwsInterruptedException{// 引用finalReentrantLock lock =this.lock;// 可中断的加锁
    lock.lockInterruptibly();try{// 看一下当前的数组还有容量没while(count ==0){// 没容量直接扔Condition队列等待
            notEmpty.await();}// 被唤醒后并且当前的数组有容量// 弹出队列中的数据即可returndequeue();}finally{// 解锁
        lock.unlock();}}

四、流程图

私聊我获取高清流程图

五、总结

鲁迅先生曾说：独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。

其实很多时候，并不是我们不够努力，很可能就是自己努力的方向不对，如果有一个人能稍微指点你一下，你真的可能会少走几年弯路。

如果你也对 后端架构和中间件源码 有兴趣，欢迎添加博主微信：hls1793929520，一起学习，一起成长

我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，喜欢后端架构和中间件源码。

我们下期再见。

我从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。

往期文章推荐：

• 从根上剖析ReentrantLock的来龙去脉
• 阅读完synchronized和ReentrantLock的源码后，我竟发现其完全相似
• 从源码全面解析 ThreadLocal 关键字的来龙去脉
• 从源码全面解析 synchronized 关键字的来龙去脉
• 阿里面试官让我讲讲volatile，我直接从HotSpot开始讲起，一套组合拳拿下面试