痛失网易30K之二：看你牛逼轰轰，请写一个阻塞队列

快来打我* 2024-03-22 21:29 33阅读 0赞

### 说在前面 ###

在40岁老架构师 尼恩的**读者社区**(50+)中，最近有小伙伴拿到了一线互联网企业如网易、极兔、有赞、希音、百度、美团的面试资格，遇到2个很重要的面试题：

> 第一弹：[为啥要用阻塞队列，用list不行吗？][list]
> 
> 第二弹：手写一个阻塞队列

阻塞队列，是面试的绝对重点和难点。

小伙伴 第一弹没有回答好，面试官又来了第二弹要求手写一个阻塞队列，**又没有写出来**…

小伙伴和尼恩说，阻塞队列虽然天天用，但是怎么实现， 还真没想过，还是要求手写…，当时就懵逼了

![d0bb5c0b29ad4b4e958caedeb572fca9.png][]

于是30K的优质offer，白白就溜走了。

![114a5e3cc081422f8ca3fbc8dc61e608.png][]

为了让后面的小伙伴不在同一个地方躺坑。

这里尼恩给大家做一下系统化、体系化的线程池梳理，使得大家可以充分展示一下大家雄厚的 “技术肌肉”，**让面试官爱到 “不能自已、口水直流”**。

也一并把这个题目以及参考答案，收入咱们的 《[尼恩Java面试宝典][Java]》V91版本，供后面的小伙伴参考，提升大家的 3高 架构、设计、开发水平。

> 注：本文以 PDF 持续更新，最新尼恩 架构笔记、面试题 的PDF文件，请到公号【技术自由圈】获取

#### 文章目录 ####

*   *  说在前面
     *  1、什么是阻塞队列？
     *  2、阻塞队列的作用
     *  3、阻塞队列的核心方法
     *   *  3.1 take 方法
         *  3.2 put 方法
     *  4、手写模拟实现一个阻塞队列
     *   *  4.1 用数组实现队列
         *  4.2 使用 synchronized 实现
         *  4.3 使用 ReentrantLock
     *  5、拆解ArrayBlockingQueue实现步骤
     *   *  5.1 用数组实现队列
         *  5.2 实现条件阻塞和线程安全
         *  5.3 put方法的流程为
         *  5.4 take方法的流程为
     *  尼恩提示
     *  说在后面
     *  推荐相关阅读

### 1、什么是阻塞队列？ ###

阻塞队列是一种队列，阻塞队列是一种特殊的队列。

阻塞队列是一种可以在多线程环境下使用，并且支持阻塞等待的队列。

![b5b326652be341bc80013ccdf6e12acc.png][]

线程 1 往阻塞队列中添加元素，当阻塞队列是满的，线程 1就会阻塞，直到队列不满

线程 2 从阻塞队列中移除元素，当阻塞队列是空的，线程 2 会阻塞，直到队列不空；

### 2、阻塞队列的作用 ###

阻塞队列，也就是 BlockingQueue，它是一个接口，如代码所示：

public interface BlockingQueue<E> extends Queue<E>{
        ...}

BlockingQueue 继承了 Queue 接口，是队列的一种。

Queue 和 BlockingQueue 都是在 Java 5 中加入的。

BlockingQueue 是线程安全的，在很多场景下都可以利用线程安全的队列来优雅地**解决业务自身的线程安全问题。**

比如说，使用生产者/消费者模式的时候，生产者只需要往队列里添加元素，而消费者只需要从队列里取出它们就可以了，如图所示：

![dc7e6266b3274ddf84e3c48f1c361d67.png][]

阻塞队列区别于其他类型的队列的最主要的特点就是“阻塞”这两个字，

**阻塞功能使得生产者和消费者两端的能力得以平衡，当有任何一端速度过快时，阻塞队列便会把过快的速度给降下来。**

### 3、阻塞队列的核心方法 ###

<table> 
 <thead> 
  <tr> 
   <th>方法类型</th> 
   <th>抛出异常</th> 
   <th>特殊值</th> 
   <th>阻塞</th> 
   <th>超时</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>插入</td> 
   <td>add(e)</td> 
   <td>offer(e)</td> 
   <td>put(e)</td> 
   <td>offer(e,time,unit)</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>remove()</td> 
   <td>poll()</td> 
   <td>take()</td> 
   <td>poll(time,unit)</td> 
  </tr> 
  <tr> 
   <td>检查</td> 
   <td>element</td> 
   <td>peek</td> 
   <td>不可用</td> 
   <td>不可用</td> 
  </tr> 
 </tbody> 
</table>

1、抛异常的方法 就是在插入满了之后，会报一个异常，remove一样，element是检查队头的元素或者是否为空。  
2、特殊值的方法是在插入满之后返回值变成了false而不是一个异常，取出失败的时候返回null。  
3、阻塞方法是在插入满之后把这个方法阻塞，一直等待队列空出来一个之后再进行加入，会出现一直等待，也可能出现饥饿现象。  
4、超时方法的话，当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程会退出。

实现阻塞最重要的两个方法是 take 方法和 put 方法。

#### 3.1 take 方法 ####

take 方法的功能是获取并移除队列的头结点，通常在队列里有数据的时候是可以正常移除的。

可是一旦执行 take 方法的时候，队列里无数据，则阻塞，直到队列里有数据。

一旦队列里有数据了，就会立刻解除阻塞状态，并且取到数据。

过程如图所示：

![f860efcbdda942a3b79afd356d149157.png][]

#### 3.2 put 方法 ####

put 方法插入元素时，如果队列没有满，那就和普通的插入一样是正常的插入，但是如果队列已满，那么就无法继续插入，则阻塞，直到队列里有了空闲空间。

如果后续队列有了空闲空间，比如消费者消费了一个元素，那么此时队列就会解除阻塞状态，并把需要添加的数据添加到队列中。

put 过程如图所示：

![42992007266644e88c18531c382e8fbb.png][]

以上过程中的阻塞和解除阻塞，都是 BlockingQueue 完成的，不需要我们自己处理。

### 4、手写模拟实现一个阻塞队列 ###

手写模拟实现一个阻塞队列，可以基于数组实现的阻塞队列，如何手写呢？

我们先从功能设计开始：

1.  首先它是一个队列，队列需要具备入队、出队的能力， 所以，设计两个方法 put、take
2.  put操作的时候，需要在队列已满时，对入队的请求进行阻塞，当队列有剩余空间时，释放入队请求；
3.  take操作的时候，在队列为空时，需要对出队的请求进行阻塞，当队列中有元素时，释放出队请求；
4.  由于ArrayBlockingQueue是一个在多线程情况下使用的数据结构，需要保证它的操作的线程安全性，所以，这里需要用到锁

#### 4.1 用数组实现队列 ####

如何用数组实现数据的入队出队操作呢？

**如何写入呢**？

这个简单，可以通过一个index字段存储当前数组下一个写入的位置。

**如何处理出队呢**?

一种简单的方法 ：简单的返回数组第一个元素，并且把后面所有的元素向前移动一位。

如果这么操作，出队时会移动大量的元素，它的时间复杂度是O(n)。

**那有没有更高效的方案呢**？

还有另一个循环数组的方案，我们通过两个int字段，分别记录下一个要入队和下一个要出队的元素的位置，当入队到数组末尾时，从0开始，同样当出队到末尾时，也从0开始。

另外当队列为空和队列已满的时候，takeIndex和putIndex都指向相同的位置，所以为了进行区分，我们可以用一个count字段存储队列元素数量，这样当count=0的时候说明队列为0，count=数组容量的时候说明队列已满

![d936a98a5b634c21884576c2e515158a.png][]

#### 4.2 使用 synchronized 实现 ####

由于 synchronized 是同一把锁，所以使用 notify() 可能会唤醒非目标线程，notifyAll() 唤醒全部线程则会带来大量的 CPU 上下文交换和锁竞争

package com.crazymakercircle.queue;
    
    public class ArrayBlockingQueue{
        
    	private Object[] array;		//数组
    	private int takeIndex;			//头
    	private int putIndex;			//尾
    	private volatile int count;	//元素个数
    
    	public ArrayBlockingQueue(int capacity){
        
    		this.array = new Object[capacity];
    	}
    	
    	//写入元素
    	public synchronized void put(Object o) throws InterruptedException{
        
    		//当队列满时，阻塞
    		while(count == array.length){
        
    			this.wait();
    		}
    		array[putIndex++] = o;
    		if(putIndex ==array.length){
        
    			putIndex = 0;
    		}
    		count++;
    		//唤醒线程
    		this.notifyAll();
    	}
    
    	//取出元素
    	public synchronized Object take() throws InterruptedException{
        
    		//当队列为空，阻塞
    		while(count == 0){
        
    			this.wait();
    		}
    		Object o = array[takeIndex++];
    		if(takeIndex == array.length){
        
    			takeIndex = 0;
    		}
    		count--;
    		//唤醒线程
    		this.notifyAll();
    		return o;
    	}
    }

#### 4.3 使用 ReentrantLock ####

可以使用 Condition 指定要唤醒的线程，所以效率高

package com.crazymakercircle.queue;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class ArrayBlockingQueueReentrantLock{
        
    	private Object[] array;		//数组
    	private int takeIndex;			//头
    	private int putIndex;			//尾
    	private volatile int count;	//元素个数
    	private ReentrantLock lock = new ReentrantLock();	//锁
    	private Condition notEmpty = lock.newCondition();	//非空
    	private Condition notFull = lock.newCondition();	//非满
    
    	public ArrayBlockingQueueReentrantLock(int capacity){
        
    		this.array = new Object[capacity];
    	}
    
    	//写入元素
    	public void put(Object o) throws InterruptedException{
        
    		try{
        
    			lock.lock();
    			//当队列满时，阻塞
    			while(count == array.length){
        
    				notFull.wait();
    			}
    			array[putIndex++] = o;
    			if(putIndex == array.length){
        
    				putIndex = 0;
    			}
    			count++;
    			//唤醒线程
    			notEmpty.notifyAll();
    		}finally{
        
    			lock.unlock();
    		}
    	}
    	
    	//取出元素
    	public Object take() throws InterruptedException{
        
    		 lock.lock();
    		 try{
        
    		 	//当队列为空，阻塞
    		 	while(count == 0){
        
    		 		notEmpty.wait();
    		 	}
    		 	Object o = array[takeIndex++];
    		 	if(takeIndex == array.length){
        
    		 		takeIndex = 0;
    		 	}
    		 	count--;
    		 	//唤醒线程
    		 	notFull.notifyAll();
    		 	return o;
    		 }finally{
        
    		 	lock.unlock();
    		 }
    	}
    }

最终，咱们要回到源码

接下来，拆解JUC源码中，ArrayBlockingQueue的实现步骤

### 5、拆解ArrayBlockingQueue实现步骤 ###

我们先拆解一下问题，把拆解ArrayBlockingQueue实现步骤分成两个步骤

1.  用数组实现队列
2.  给队列加上阻塞能力和保证线程安全

#### 5.1 用数组实现队列 ####

使用 takeIndex、putIndex 避免数组复制

下面代码展示了用数组实现队列的具体实现。

class ArrayBlockingQueue<E> {
        
        final Object[] items;
        int takeIndex;
        int putIndex;
        int count;
        public ArrayBlockingQueue(int capacity) {
        
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
        }
        private void enqueue(E e) {
        
            Object[] items = this.items;
            items[putIndex] = e;
            if (++putIndex == items.length) putIndex = 0;
            count++;
        }
        private E dequeue() {
        
            Object[] items = this.items;
            E e = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length) takeIndex = 0;
            count--;
            return e;
        }
    }

#### 5.2 实现条件阻塞和线程安全 ####

![8d9244fc71824489aaa16468c163c2e1.png][]

「在队列已满时，对入队的请求进行阻塞，当队列有剩余空间时，释放入队请求」这个需求本质上是一个条件等待的特例，写入的条件是队列不满，不满足条件的时候需要等待，直到满足条件为止。

在Java中，实现条件等待有synchronized+Object.wait和Lock+Condition.await两种方式，这里不用synchronized方案，是因为

1.  synchronized不支持interrupt
2.  synchronized无法支持多个条件

通过Lock和Condition的方案，还能够保证线程安全，因为上面的环形数组实现中，线程间共享的变量有items数组、takeIndex、putIndex、count，线程安全涉及到原子性可见性重排序几个方面，通过Lock类加锁可以对共享变量的读写操作进行保护。

定义阻塞的Lock对象和Condition，条件分为不满和不空两个条件。

class ArrayBlockingQueue<E> {
        
        final Object[] items;
        int takeIndex;
        int putIndex;
        int count;
        ReentrantLock lock;
        private final Condition notEmpty;
        private final Condition notFull;
        public ArrayBlockingQueue(int capacity) {
        
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
            // 创建lock对象
            lock = new ReentrantLock();
            // 创建非空的Condition
            notEmpty = lock.newCondition();
            // 创建不满的Condition
            notFull =  lock.newCondition();
        }
    }

以入队操作添加实现为例，能够入队的条件是队列不满，也就是count < items.length，不能入队的条件反过来就是count == items.length。

当满足条件后，我们就可以入队了，入队之后，还需要唤醒等待出队的线程。

#### 5.3 put方法的流程为 ####

1.  先加锁
2.  在锁中while循环判断条件是否满足，不满足调用notFull.await()，await()方法会释放锁，被其他线程signal唤醒后会重新抢锁，再次获得锁后会继续走到while循环判断条件的地方。
3.  如果条件已经满足，则执行入队操作
4.  入队完之后调用notEmpty.signal()唤醒一个等待notFull条件的线程
5.  finally中释放锁

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

方法中还有一些小细节

1.  put方法中，为什么要先用一个声明一个lock局部变量呢？

ReentrantLock lock = this.lock;

这是因为如果不使用局部变量，后面所有使用实例变量的调用，在字节码指令层面需要变成先调用aload 0获取到this，再调用getField指令获取字段值，再进行其他操作。

而先把lock存到局部变量中，后面所有的获取lock就可以变成一个aload xxx指令，从而节省了指令数量，也就会加快方法的执行速度。

1.  为什么while循环需要放在锁内呢？

如果不放在锁内，则可能会出现多个线程同时看到满足条件，进而去加锁入队。

虽然入队还是在临界区，但是会出现队列已满，仍然在执行入队操作的情况。

这个问题和单例的double check locking中少些一个check的问题类似。

#### 5.4 take方法的流程为 ####

take方法是和put相对应的出队方法，和put流程基本一致

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

### 尼恩提示 ###

更多的 JUC 高并发知识，请参见《[Java 高并发核心编程 卷2 加强版][Java _ _2]》PDF

### 说在后面 ###

如果手写到后面一个版本，并且能把 **实例变量的调用的性能优化**，**while循环为何要放在锁内**等这些高超的技术点写出来，那么太牛了。

那么面试官一定将你归为**技术大佬、技术高手**，面试官已经爱到 “不能自已、口水直流” 啦。

**offer，当然也就来了。**

此真题收入4800页《[尼恩Java面试宝典][Java]》V91版，最新的PDF找尼恩获取。

![84a4392554954ec4b69a0ea60bcf68a6.png][]

### 推荐相关阅读 ###

《[痛失网易30K之一：为啥用阻塞队列，list不行吗？][list]》

《[腾讯太狠：10亿QPS的IM，如何实现？][10_QPS_IM]》

《[美团二面：epoll性能那么高，为什么？][epoll]》

《[腾讯太狠：40亿QQ号，给1G内存，怎么去重？][40_QQ_1G]》

《[京东太狠：100W数据去重，用distinct还是group by，说说理由？][100W_distinct_group by]》

《[美团太狠：接口被恶刷10Wqps，怎么防？][10Wqps]》

《[顶奢好文：3W字，穿透Spring事务原理、源码，最少读10遍][3W_Spring_10]》

> 《尼恩 架构笔记》《尼恩高并发三部曲》《尼恩Java面试宝典》PDF，请到下面公号【技术自由圈】取↓↓↓

[list]: https://blog.csdn.net/crazymakercircle/article/details/131887321
[d0bb5c0b29ad4b4e958caedeb572fca9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/3be4036af9db4f92acb3cbda3443ada8.png
[114a5e3cc081422f8ca3fbc8dc61e608.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/f7c55434165a4eb295c0a57af5b2a11c.png
[Java]: https://blog.csdn.net/crazymakercircle/article/details/124790425
[b5b326652be341bc80013ccdf6e12acc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/27f38120b79443e681d1b163b3cfd356.png
[dc7e6266b3274ddf84e3c48f1c361d67.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/d8710c322b1049d9a4407c2f796e3f91.png
[f860efcbdda942a3b79afd356d149157.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/c02b91e8c2c24957beccd488e63e3e48.png
[42992007266644e88c18531c382e8fbb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/6d574bd7cacd4583827ec237c541a7b3.png
[d936a98a5b634c21884576c2e515158a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/21503e7671b44ce38c5908c7b536b9e2.png
[8d9244fc71824489aaa16468c163c2e1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/f39631c03a654ba2a8127d35ab4c9ca1.png
[Java _ _2]: https://blog.csdn.net/crazymakercircle/article/details/115677785
[84a4392554954ec4b69a0ea60bcf68a6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/db650dd938544d97980c88484b27454d.png
[10_QPS_IM]: https://blog.csdn.net/crazymakercircle/article/details/131936651
[epoll]: https://blog.csdn.net/crazymakercircle/article/details/131887110
[40_QQ_1G]: https://blog.csdn.net/crazymakercircle/article/details/131256900
[100W_distinct_group by]: https://blog.csdn.net/crazymakercircle/article/details/131030436
[10Wqps]: https://blog.csdn.net/crazymakercircle/article/details/131067131
[3W_Spring_10]: https://blog.csdn.net/crazymakercircle/article/details/131214847