jdk11源码--ArrayBlockingQueue源码分析

逃离我推掉我的手 2021-12-18 19:15 423阅读 0赞

> 更多java源码分析请见：jdk11源码分析系列文章专栏：[Java11源码分析][Java11]

### 文章目录 ###

*  欢迎关注本人公众号
 *  概述
 *  BlockingQueue
 *  ArrayBlockingQueue
 *   *  ArrayBlockingQueue构造方法
     *  put
     *  take
 *  总结

# 欢迎关注本人公众号 #

![在这里插入图片描述][20200515113638671.jpg_pic_center]

# 概述 #

上一篇文章[jdk11源码–ReentrantLock之Condition源码分析][jdk11_ReentrantLock_Condition]中分析了ReentrantLock和Condition的源码，那么接下来看一下Condition在JDK中的具体应用。

ArrayBlockingQueue底层就是使用Condition来实现的。

# BlockingQueue #

BlockingQueue 阻塞队列，该类是一个接口，平时我们熟知的ArrayBlockingQueue，LinkedBlockingQueue等都是该接口的实现。  
BlockingQueue 之所以说是阻塞的，是因为他可以在队列为空的时候，获取元素的线程会阻塞，直到有新的元素添加进来。当队列满时，添加元素的线程会阻塞，直到有线程从队列中取走了元素。这也是注明的生产者消费者问题。  
当有面试官问你生产者消费者问题时，直接将ArrayBlockingQueue源码分析讲一下也是可以的。

看一下BlockingQueue的类图，BlockingQueue是继承自Queue，而queue继承自collection。  
![在这里插入图片描述][20190701144758159.png]

BlockingQueue 对插入、删除、获取原色的操作提供了四种不同的方法用于不同的场景中使用，这些方法总结在下表中：

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>抛出异常Throws exception</th> 
   <th>Special value</th> 
   <th>Blocks</th> 
   <th>Times out</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>插入数据Insert</td> 
   <td>add(e) //队列满时抛异常</td> 
   <td>offer(e) //队列满返回false，不阻塞</td> 
   <td><strong>put(e)</strong> //队列满时阻塞，直到队列未满时再插入</td> 
   <td>offer(e, time, unit) //指定直接内可以插入返回true，指定时间内不能插入，返回false</td> 
  </tr> 
  <tr> 
   <td>获取数据Remove</td> 
   <td>remove()//队列为空，抛异常</td> 
   <td>poll()//队列为空返回null，不阻塞</td> 
   <td><strong>take()</strong> //当队列为空时会阻塞，一直等到队列不为空时再返回队首值</td> 
   <td>poll(time, unit) //在指定时间内，队列都是空，则返回null，否则返回对首的值</td> 
  </tr> 
  <tr> 
   <td>Examine</td> 
   <td>element()</td> 
   <td>peek()</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

本文我们重点关注 put和take方法，因为这两个方法时阻塞的。

# ArrayBlockingQueue #

ArrayBlockingQueue是BlockingQueue的一个实现。他是FIFO先进先出队列。  
ArrayBlockingQueue类图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]  
重要属性：

/** 队列集合，数组存储！！ */
    final Object[] items;
    /** take, poll, peek or remove 方法获取元素的下标位置 */
    int takeIndex;
    /** put, offer, or add 方法添加元素的下标位置 */
    int putIndex;
    
    /** 队列中的元素数量 */
    int count;
    
    //并发控制
    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

items是使用数组存储的，这也是ArrayBlockingQueue名称的由来。

并发控制使用经典的双Condition 算法，上面定义了两个Condition ，一个notEmpty，一个notFull。下面来逐行源码具体分析一下。

## ArrayBlockingQueue构造方法 ##

//capacity: 数组初始容量
    public ArrayBlockingQueue(int capacity) { 
        this(capacity, false);
    }
    //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();
    }

在构造函数中会指定初始容量以及锁的类型，默认是非公平锁。  
数组items一旦确定下来，后续就不会再更改大小。

## put ##

put方法添加元素到队尾，当队列满时阻塞。

public void put(E e) throws InterruptedException { 
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;//获取锁lock
        lock.lockInterruptibly();//加锁，响应中断
        try { 
            while (count == items.length)
                notFull.await();//如果items满了，那么notFull阻塞
            enqueue(e);//将元素添加到队列末尾
        } finally { 
            lock.unlock();
        }
    }
    
    //将元素添加到队列末尾, ++putIndex,count++,
    //该方法只允许在lock加锁后操作
    private void enqueue(E e) { 
        final Object[] items = this.items;
        items[putIndex] = e;
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();//唤醒阻塞的获取元素的线程
    }

整个过程还是比较简单的，首先加锁，注意这里的锁允许中断返回。然后，如果队列满了（`count == items.length`），那么就无法继续添加元素了，添加元素的线程就需要await等待（`notFull.await();`），该线程添加到notFull的condition队列上，直到被take方法唤醒（后面会讲）后，继续添加元素到队尾（`enqueue(e);`）。  
enqueue方法中，添加元素到putIndex 的位置，然后对putIndex 加1操作，由于是数组，所以这里进行了越界处理，越界后从0开始继续计算。count元素的数量进行加1操作，同时唤醒notEmpty condition队列上阻塞的线程。  
读者可以思考一下这里为什么没有进行这个校验：putIndex 在加一以后是否会与现有队列中已经存在的元素重合而覆盖掉现有元素？答案下一节揭晓。

## take ##

take：从队首获取元素。

public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();//加锁，响应中断
        try { 
            while (count == 0)
                notEmpty.await();//如果队列中空，那么当前线程需要添加到notEmpty的condition队列中阻塞，直到有新的元素添加进来
            return dequeue();
        } finally { 
            lock.unlock();
        }
    }
    
    //从对首获取元素，
    //该方法只允许在lock加锁后操作
    private E dequeue() { 
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E e = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length) takeIndex = 0;//设置takeIndex 下一次应该获取的位置
        count--;//队列总数量-1
        if (itrs != null)
            itrs.elementDequeued();//迭代器相关
        notFull.signal();//唤醒notFull condition队列上的线程
        return e;
    }

take过程也很简单，先看一下队列是否为空，如果为空，则无法获取元素，将当前线程添加到notEmpty的condition队列。否则从takeIndex 的位置读取元素并且设置takeIndex ，count 的值。

上一节提到了这个问题：  
putIndex 在加一以后是否会与现有队列中已经存在的元素重合而覆盖掉现有元素？  
其实也很简单，enqueue和dequeue都是需要加锁以后才调用的，所以是线程安全的，count可以准确表示队列中的有效长度，takeIndex 和putIndex 也都没有并发问题，他们每次添加或者读取时都会判断count的值，来确认队列是否满或者空。如此一来，当然不会出现添加过多覆盖现有元素的情况。

# 总结 #

首先回顾一下上一篇[jdk11源码–ReentrantLock之Condition源码分析][jdk11_ReentrantLock_Condition]中画的condition结构图吗，以及condition与ReentrantLock之间的关系。  
然后再次基础上，画一下ArrayBlockingQueue中的condition关系图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]  
总体来讲，ArrayBlockingQueue包含一个ReentrantLock和两个condition：notEmpty和notFull。  
put可take操作都需要加锁，都是线程安全的。  
当队列满时，put操作需阻塞等待，当前线程添加到notFull的condition队列中；  
当队列不满，添加元素成功时，需唤醒notEmpty队列中的线程；  
当队列空时，take操作需阻塞等待，当前线程添加到notEmpty的condition队列中；  
当队列不空，获取元素成功时，需唤醒notFull队列中的线程。

[Java11]: https://it007.blog.csdn.net/column/info/30895
[20200515113638671.jpg_pic_center]: /images/20211219/be5d5f05a2ee4d06a757bee939f7326b.png
[jdk11_ReentrantLock_Condition]: https://blog.csdn.net/liubenlong007/article/details/93863136
[20190701144758159.png]: /images/20211219/41712d2d1e1d464db0834d498b5af117.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: /images/20211219/32e5879dd7d0486ca99e51ad3e0d3c55.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: /images/20211219/33b1c827ceb34876a3640e23e280c878.png