阻塞队列详解 ArrayBlockingQueue 源码分析 -- 基于JDK 1.8

我就是我 2021-11-14 13:20 280阅读 0赞

阻塞队列产生前景：在实际应用场景中，多生产者、多消费者模式，如果我们自己构造一个FIFO先进先出队列，在多线程从队列存取数据的时候，需要操作一些临界资源，我们不仅要考虑加锁的情况，还要考虑队列是否满了，满了就要wait直到队列不满的时候再往里面存，并且唤醒notify取元素线程，相反，队列空了就wait直到队列不空的时候再从里面取，唤醒notify存元素线程；一但操作不当，容易造成阻塞唤醒问题，容易引起死锁；所以Concurrent包的作者为我们编写了各种阻塞队列；

下面将要介绍的是阻塞队列之一的ArrayBlockingQueue：

**目录**

1、ArrayBlockingQueue简介

2、ArrayBlockingQueue结构

3、类中的变量和构造方法

基本结构：

类中的属性：

构造方法如下：

4、插入元素方法

（1）非阻塞方法add（调用offer）

（2）阻塞方法put，存放元素

5、删除元素方法

（1）、非阻塞poll方法，移除头元素

（2）、非阻塞remove方法，移除指定元素

（3）、阻塞take方法，移除头元素

6、返回队列头元素方法peek（不删除）

（1）、直接返回头元素，但是并不删除头元素

7、其他方法，不一一分析

--------------------

# 1、ArrayBlockingQueue简介 #

ArrayBlockingQueue是属于一种FIFO先进先出阻塞队列，一个底层由数组结构组成的有界阻塞队列；为什么叫阻塞队列（BlockingQueue）？比普通队列多了两个阻塞操作的方法，即阻塞的插入和阻塞移除：

（1）、支持阻塞的插入方法put()：添加元素时，当阻塞队列元素已满时（前提是有界队列），该线程不会抛出异常，线程阻塞，等到元素不满时才重新唤醒线程执行元素加入操作；

（2）、支持阻塞的移除方法take()：删除时，在队列元素为空时，删除队列元素的线程不会抛出异常，将被阻塞，等到队列有元素时再执行删除操作；

# 2、ArrayBlockingQueue结构 #

我们要讲解的ArrayBlockingQueue结构如下图，在IDEA工具中，先找到类源码，使用Ctrl+Shift+Alt+U 快捷键，即可看到源码的结构图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70][]

# 3、类中的变量和构造方法 #

## 基本结构： ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1][]

## 类中的属性： ##

/**
     * Serialization ID. 用于序列化
     */
    private static final long serialVersionUID = -817911632652898426L;
    
    /** 存放的元素，说明底层是数组结构 */
    final Object[] items;
    
    /**元素的取出索引，用于下一次调用如下方法时使用：
    
    take, poll, peek or remove ；
    
    队列中没有存整个数组的下标索引*/
    int takeIndex;
    
    /** 元素的添加索引，用于下一次调用如下方法时使用：
    
    put, offer, or add ；
    
    队列中没有存整个数组的下标索引*/
    int putIndex;
    
    /** 队列中元素个数 */
    int count;
    
    /** 主要的锁，用来保护所有的访问*/
    final ReentrantLock lock;
    
    /** 等待获取条件Condition for waiting takes */
    private final Condition notEmpty;
    
    /**等待存放条件 Condition for waiting puts */
    private final Condition notFull;
    
    /**
    迭代器
     */
    transient Itrs itrs = null;

## 构造方法如下： ##

/**默认构造方法 */
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }
    
    /**各项初始化工作，默认是非公平阻塞队列；
    公平与非公平，底层是使用Reentrantlock的公平与非公平锁来实现；
    所以非公平方式，效率更高一些，在前面分析Reentrantlock有提到；
    什么是公平队列，被阻塞的线程可以按照阻塞的先后顺序访问队列，
    即先阻塞的线程先访问队列。而非公平队列，当队列可用时，阻塞
    的线程将进入争夺访问资源的竞争中，也就是说谁先抢到谁就执行，
    没有固定的先后顺序；*/
    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();
    }

# 4、插入元素方法 #

## （1）非阻塞方法add（调用offer） ##

添加失败抛出IllegalStateException异常

public boolean add(E e) {
    
          //调用父类的add方法
         return super.add(e);
    }
    
    
    
    //父类AbstractQueue中的add方法
    
    public boolean add(E e) {
    
        //如果添加成功，返回true，否则抛出IllegalStateException
    
        //这时调用的是子类ArrayBlockingQueue中的offer方法
    
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
    
    
    
    //子类ArrayBlockingQueue中的offer方法
    public boolean offer(E e) {
    
        //检查是否为空
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
    
        //获取锁
    
        lock.lock();
        try {
    
        //如果队列满了，那么返回false，添加失败；
    
        //到父类中的add方法，false就会抛出IllegalStateException
    
          if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
    
          //在finally块中释放锁
            lock.unlock();
        }
    }
    
    /** 内部私有方法*/
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
    
          //如果数组元素满了以后，会重置存放索引为0
    
          //为什么会重置为0，因为队列满了以后，
    
          //再次添加元素的时候，会从队列头部开始，
    
          //如果满队列此时没有元素出队，添加元素要么失败抛出异常，
    
          //要么阻塞
          if (++putIndex == items.length)
            putIndex = 0;
            count++;
       
    
          //唤醒在等待获取元素的线程
    
          notEmpty.signal();
    }

//此外还有带超时的offer方法，
    
    //规定时间内添加元素，否则超时抛出异常
    
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException

## （2）阻塞方法put，存放元素 ##

/** put函数用于存放元素，在当前线程被中断时会抛出异常/
    public void put(E e) throws InterruptedException {
    
        //检查是否为空
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
    
        //获取，可中断锁，如果未被中断，则线程持有锁
        lock.lockInterruptibly();
        try {
    
            //如果元素满了，则采用condition等待，
            //notFull添加元素条件进行等待；
            while (count == items.length)
            notFull.await();
            //否则正常添加
            enqueue(e);
        } finally {
    
            //在finally块中释放锁
            lock.unlock();
        }
    }

# 5、删除元素方法 #

## （1）、非阻塞poll方法，移除头元素 ##

获取并移除此队列的头元素，若队列为空，则返回 null

public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
             //如果队列为空，返回null
             //否则调用dequeue方法获取元素
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }
    
    
    
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
    
        //因为需要取出这个元素，所以置空该获取索引
        items[takeIndex] = null;
    
        //元素空了以后，重置获取索引为0；
    
        //原理跟上面putIndex重置为0一样
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
    
          //当数组元素为空，或者获取下标重置0时候，需要更新迭代器
          itrs.elementDequeued();
          notFull.signal();
          //返回被删除的元素
    
        return x;
    }

//此外还有带超时的poll方法，
    
    //规定时间内移除返回元素，否则超时抛出异常
    
    public E poll(long timeout, TimeUnit unit) throws InterruptedException

## （2）、非阻塞remove方法，移除指定元素 ##

public boolean remove(Object o) {
    
        if (o == null) return false;
    
        final Object[] items = this.items;
    
        final ReentrantLock lock = this.lock;
    
        lock.lock();
    
        try {
    
            if (count > 0) {
    
                final int putIndex = this.putIndex;
                //获取下标
    
                int i = takeIndex;
    
                do {
                    //循环，一直找到要删除的元素
                    if (o.equals(items[i])) {
    
                        removeAt(i);
    
                        return true;
    
                    }
    
                    if (++i == items.length)
    
                        i = 0;
    
                } while (i != putIndex);
    
            }
    
            return false;
    
        } finally {
    
            lock.unlock();
    
        }
    
    }
    
    void removeAt(final int removeIndex) {
    
        // assert lock.getHoldCount() == 1;
    
        // assert items[removeIndex] != null;
    
        // assert removeIndex >= 0 && removeIndex < items.length;
    
        final Object[] items = this.items;
    
        if (removeIndex == takeIndex) {
    
            // removing front item; just advance
    
            items[takeIndex] = null;
                  //元素空了以后，重置获取下标索引
    
            if (++takeIndex == items.length)
    
                takeIndex = 0;
    
            count--;
    
            if (itrs != null)
    
                itrs.elementDequeued();
    
        } else {
    
            // an "interior" remove
    
    
    
            // slide over all others up through putIndex.
    
            final int putIndex = this.putIndex;
    
            for (int i = removeIndex;;) {
    
                int next = i + 1;
    
                if (next == items.length)
    
                    next = 0;
    
                if (next != putIndex) {
    
                    items[i] = items[next];
    
                    i = next;
    
                } else {
    
                    items[i] = null;
    
                    this.putIndex = i;
    
                    break;
    
                }
    
            }
    
            count--;
    
            if (itrs != null)
    
                itrs.removedAt(removeIndex);
    
        }
                  //唤醒等待添加的阻塞队列
    
        notFull.signal();
    
    }

## （3）、阻塞take方法，移除头元素 ##

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
    
                //删除以后，唤醒添加阻塞的线程
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
    
    //逻辑跟入队列相似
    
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }

这里有一个小tip就是：take和put方法，都是阻塞的，都带字母t，我们可以这样去记；

# 6、返回队列头元素方法peek（不删除） #

## （1）、直接返回头元素，但是并不删除头元素 ##

public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return itemAt(takeIndex); // null when queue is empty
        } finally {
            lock.unlock();
        }
    }

# 7、其他方法，不一一分析 #

//返回数组队列大小
    
    public int size()
    
    //查找是否包含元素
    
    public boolean contains(Object o)
    
    //迭代队列数组
    
    public Iterator<E> iterator() {
        return new Itr();
    }

以上是整个阻塞队列ArrayBlockingQueue的分析，不是很难，比起看ConcurrentHashMap、AQS之类的源码简单多了，花了一点时间分析一下。

[阻塞队列详解 LinkedBlockingQueue 源码分析 -- 基于JDK 1.8][LinkedBlockingQueue _ -- _JDK 1.8]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70]: /images/20211114/847b5ea34c1149dd9b3e237aa00e15bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1]: /images/20211114/54bc4606fab447adac6cb2f2f3d77edc.png
[LinkedBlockingQueue _ -- _JDK 1.8]: https://blog.csdn.net/lejustdoit/article/details/98107435