ArrayBlockingQueue源码详解

墨蓝 2022-10-31 00:58 172阅读 0赞

## 阻塞队列 ##

*  插入方法：
    
     *  add(E e) : 添加成功返回true，失败抛IllegalStateException异常
     *  offer(E e) : 成功返回 true，如果此队列已满，则返回 false。
     *  put(E e) :将元素插入此队列的尾部，如果该队列已满，则一直阻塞
 *  删除方法:
    
     *  remove(Object o) :移除指定元素,成功返回true，失败返回false
     *  poll() : 获取并移除此队列的头元素，若队列为空，则返回 null
     *  take()：获取并移除此队列头元素，若没有元素则一直阻塞。
 *  检查方法
    
     *  element() ：获取但不移除此队列的头元素，没有元素则抛异常
     *  peek() :获取但不移除此队列的头；若队列为空，则返回 null。

**ArrayBlockingQueue原理概要**  
ArrayBlockingQueue的内部是通过一个可重入锁ReentrantLock和两个Condition条件对象来实现阻塞，这里先看看其内部成员变量

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable { 
    
        /** 存储数据的数组 */
        final Object[] items;
    
        /**获取数据的索引，主要用于take，poll，peek，remove方法 */
        int takeIndex;
    
        /**添加数据的索引，主要用于 put, offer, or add 方法*/
        int putIndex;
    
        /** 队列元素的个数 */
        int count;
    
    
        /** 控制并非访问的锁 */
        final ReentrantLock lock;
    
        /**notEmpty条件对象，用于通知take方法队列已有元素，可执行获取操作 */
        private final Condition notEmpty;
    
        /**notFull条件对象，用于通知put方法队列未满，可执行添加操作 */
        private final Condition notFull;
    
        /** 迭代器 */
        transient Itrs itrs = null;
    
    }

从成员变量可看出，ArrayBlockingQueue内部确实是通过数组对象items来存储所有的数据，值得注意的是ArrayBlockingQueue通过一个ReentrantLock来同时控制添加线程与移除线程的并非访问，这点与LinkedBlockingQueue区别很大(稍后会分析)。而对于notEmpty条件对象则是用于存放等待或唤醒调用take方法的线程，告诉他们队列已有元素，可以执行获取操作。同理notFull条件对象是用于等待或唤醒调用put方法的线程，告诉它们，队列未满，可以执行添加元素的操作。takeIndex代表的是下一个方法(take，poll，peek，remove)被调用时获取数组元素的索引，putIndex则代表下一个方法（put, offer, or add）被调用时元素添加到数组中的索引。图示如下  
![img][]

**ArrayBlockingQueue的(阻塞)添加的实现原理**

### enqueue ###

//add方法实现，间接调用了offer(e)
    public boolean add(E e) { 
            if (offer(e))
                return true;
            else
                throw new IllegalStateException("Queue full");
        }
    
    //offer方法
    public boolean offer(E e) { 
         checkNotNull(e);//检查元素是否为null
         final ReentrantLock lock = this.lock;
         lock.lock();//加锁
         try { 
             if (count == items.length)//判断队列是否满
                 return false;
             else { 
                 enqueue(e);//添加元素到队列
                 return true;
             }
         } finally { 
             lock.unlock();
         }
     }
    
    //入队操作
    private void enqueue(E x) { 
        //获取当前数组
        final Object[] items = this.items;
        //通过putIndex索引对数组进行赋值
        items[putIndex] = x;
        //索引自增，如果已是最后一个位置，重新设置 putIndex = 0;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;//队列中元素数量加1
        //唤醒调用take()方法的线程，执行元素获取操作。
        notEmpty.signal();
    }

这里的add方法和offer方法实现比较简单，其中需要注意的是enqueue(E x)方法，其方法内部通过putIndex索引直接将元素添加到数组items中，这里可能会疑惑的是当putIndex索引大小等于数组长度时，需要将putIndex重新设置为0，这是因为当前队列执行元素获取时总是从队列头部获取，而添加元素从中从队列尾部获取所以当队列索引（从0开始）与数组长度相等时，下次我们就需要从数组头部开始添加了，如下图演示  
![img][img 1]

ok~,接着看put方法，它是一个阻塞添加的方法，

### put ###

//put方法，阻塞时可中断
     public void put(E e) throws InterruptedException { 
         checkNotNull(e);
          final ReentrantLock lock = this.lock;
          lock.lockInterruptibly();//该方法可中断
          try { 
              //当队列元素个数与数组长度相等时，无法添加元素
              while (count == items.length)
                  //将当前调用线程挂起，添加到notFull条件队列中等待唤醒
                  notFull.await();
              enqueue(e);//如果队列没有满直接添加。。
          } finally { 
              lock.unlock();
          }
      }

put方法是一个阻塞的方法，如果队列元素已满，那么当前线程将会被notFull条件对象挂起加到等待队列中，直到队列有空档才会唤醒执行添加操作。但如果队列没有满，那么就直接调用enqueue(e)方法将元素加入到数组队列中。到此我们对三个添加方法即put，offer，add都分析完毕，其中offer，add在正常情况下都是无阻塞的添加，而put方法是阻塞添加。这就是阻塞队列的添加过程。说白了就是当队列满时通过条件对象Condtion来阻塞当前调用put方法的线程，直到线程又再次被唤醒执行。总得来说添加线程的执行存在以下两种情况，一是，队列已满，那么新到来的put线程将添加到notFull的条件队列中等待，二是，有移除线程执行移除操作，移除成功同时唤醒put线程，如下图所示  
![img][img 2]

**ArrayBlockingQueue的(阻塞)移除实现原理**

### dequeue ###

public E poll() { 
          final ReentrantLock lock = this.lock;
           lock.lock();
           try { 
               //判断队列是否为null，不为null执行dequeue()方法，否则返回null
               return (count == 0) ? null : dequeue();
           } finally { 
               lock.unlock();
           }
        }
     //删除队列头元素并返回
     private E dequeue() { 
         //拿到当前数组的数据
         final Object[] items = this.items;
          @SuppressWarnings("unchecked")
          //获取要删除的对象
          E x = (E) items[takeIndex];
          将数组中takeIndex索引位置设置为null
          items[takeIndex] = null;
          //takeIndex索引加1并判断是否与数组长度相等，
          //如果相等说明已到尽头，恢复为0
          if (++takeIndex == items.length)
              takeIndex = 0;
          count--;//队列个数减1
          if (itrs != null)
              itrs.elementDequeued();//同时更新迭代器中的元素数据
          //删除了元素说明队列有空位，唤醒notFull条件对象添加线程，执行添加操作
          notFull.signal();
          return x;
        }

### take ###

//从队列头部删除，队列没有元素就阻塞，可中断
     public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
          lock.lockInterruptibly();//中断
          try { 
              //如果队列没有元素
              while (count == 0)
                  //执行阻塞操作
                  notEmpty.await();
              return dequeue();//如果队列有元素执行删除操作
          } finally { 
              lock.unlock();
          }
        }

take方法其实很简单，有就删除没有就阻塞，注意这个阻塞是可以中断的，如果队列没有数据那么就加入notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put线程添加了数据，那么put操作将会唤醒take线程，执行take操作。

参考文章  
[https://blog.csdn.net/javazejian/article/details/77410889][https_blog.csdn.net_javazejian_article_details_77410889]

[img]: /images/20221024/03e73e70a7ce48b9a79ef16ace5bcf58.png
[img 1]: /images/20221024/981aa4a5b89541f7a7c77f7d82ef7d1a.png
[img 2]: /images/20221024/38ca163acdbd4029ab59e86c3843d223.png
[https_blog.csdn.net_javazejian_article_details_77410889]: https://blog.csdn.net/javazejian/article/details/77410889