【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析

拼搏现实的明天。 2022-12-11 02:26 359阅读 0赞

> 阻塞队列系列：
> 
>  *  [【JUC源码】阻塞队列：继承关系分析][JUC]
>  *  [【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析][JUC_ArrayBlockingQueue]
>  *  [【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析][JUC_LinkedBlockingQueue]
>  *  [【JUC源码】阻塞队列：DelayQueue 源码分析][JUC_DelayQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）][JUC_SynchronousQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）][JUC_SynchronousQueue 1]
>  *  [【JUC源码】阻塞队列：总结 --特性对比&应用场景举例][JUC_ --]
>  *  [【JUC源码】阻塞队列：各种阻塞队列面试题合集][JUC 1]

从类注释可以得到的信息如下：

1.  有界的阻塞数组，容量一旦创建，后续大小无法修改；
2.  元素是有顺序的，按照先入先出进行排序，从队尾插入数据数据，从队头拿数据；
3.  队列满时，往队列中 put 数据会被阻塞，队列空时，往队列中拿数据也会被阻塞。

ArrayBlockingQueue 的继承关系，核心成员变量及主要构造函数：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
    	implements BlockingQueue<E>, java.io.Serializable {
        
    	
    	// 保存数据的数组，ArrayBlockingQueue 实际上是通过数组实现了一个循环队列
        // 注：数组大小必须在初始化的时候手动设置，没有默认大小
        final Object[] items;
    
        // 下次拿数据的时候的索引位置（队首）
        int takeIndex;                   						  
        // 下次放数据的索引位置（队尾）
        int putIndex;
        // 队列中已有元素的个数（<=length）
        int count;
        
        /**
        * 注：takeIndex和putIndex都是动态变化的，因为数组容量（length）初始化好就始终不会再变，主要的变化为两种：
        * 1.在元素入队后putIndex++，在元素出队后takeIndex++
        * 2.在putIndex或者takeIndex == length-1（到达数组末端），且count != length（队列未满）时，它俩会被重置为0
        */
    
        // 可重入的锁，作用有两个
        // 1.保证在并发时对数组操作（入队/出队）的线程安全。另外，还可以设置是否公平锁
        // 2.通过条件队列（Condition）实现线程控制
        final ReentrantLock lock;
    
    	// 入队(put)的条件队列，其实很容易理解：入队条件是队列未满
        private final Condition notFull;
        
        // 出队(take)的条件队列。其实很容易理解：出队条件是队列非空
        private final Condition notEmpty;
    
        
        //----------------------------------构造函数--------------------------------
        // 构造函数一：传入初始容量
        // 注：ArrayBlockingQueue 无空参构造（因为循环队列无扩容操作）
        public ArrayBlockingQueue(int capacity) {
        
        	// 默认是非公平锁
            this(capacity, false);
        }
        
        // 构造函数二：传入初始容量和是否公平锁
        public ArrayBlockingQueue(int capacity, boolean fair) {
        
        	// 容量参数合法判断
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
    
            // 创建锁
            // 1.公平，即按照阻塞的先后顺序唤醒；
            // 2.非公平（默认），未阻塞或阻塞时间少的线程有机会先获得锁
            lock = new ReentrantLock(fair);
            
            // 初始化两个条件队列
            notEmpty = lock.newCondition();
            notFull =  lock.newCondition();
        }
        
        // 构造函数三：传入一个集合
        public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) {
        
            this(capacity, fair);
    
            final ReentrantLock lock = this.lock;
            lock.lock(); // Lock only for visibility, not mutual exclusion
            try {
        
                // i 代表插入的位置
                int i = 0;
                try {
        
                    // 此时需要注意的是，如果 c 的大小超过了数组的大小
                    // 是会抛异常的
                    for (E e : c) {
        
                        checkNotNull(e);
                        items[i++] = e;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) {
        
                    throw new IllegalArgumentException();
                }
                count = i;
                // 如果插入的位置，正好是队尾了，下次需要从 0 开始插入
                putIndex = (i == capacity) ? 0 : i;
            } finally {
        
                lock.unlock();
            }
        }
    }

在[阻塞队列继承关系分析][JUC]中，我们看到了队列的主要方法无非就三个：入队、出队、队首。所以接下来我们就从这三个方法入手，来看看 ArrayBlockingQueue 的具体实现。

## 1.入队 ##

### put()：队满阻塞 ###

public void put(E e) throws InterruptedException {
        
            // 入队元素不能为空
            checkNotNull(e); 
            final ReentrantLock lock = this.lock;
            // 上锁（可中断），让当前线程进入同步队列
            // 注：如果是非公平锁，则可能后来的线程直接拿锁，而不用进入同步队列
            lock.lockInterruptibly(); 
            try {
        
                // ！！！队列如果是满的，就将当前线程加入到notFull的条件队列中，然后进入阻塞状态
                // 等待某个线程执行拿走队列元素（dequeue）或者删除元素（removeAt）后，有机会被唤醒进入同步队列
                // 注：这里的while循环有double-check的意思，即防止线程已经被调度执行了，但前一刻有另一个线程put/offer/add成功了，所以他又要进入条件队列notFull中阻塞等待。
                while (count == items.length)
                    notFull.await();
                // 入队时数组的相关操作
                enqueue(e); 
            } finally {
        
            	// 释放锁
                lock.unlock(); 
            }
    }

这里面涉及到的是AQS相关源码，感兴趣的童鞋可以参考我[关于 AQS 系列的文章][AQS]…

### offer()：队满返回false ###

public boolean offer(E e) {
        
    		 // 入队元素不能为空
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            // 加锁
            lock.lock(); 
            try {
        
                // ！！！如果队列是满的，直接返回 false
                if (count == items.length)
                    return false;
                else {
        
                	// 入队时数组的相关操作
                    enqueue(e); 
                    return true;
                }
            } finally {
        
            	// 释放锁
                lock.unlock(); 
            }
    }

### add()：队满抛异常 ###

public boolean add(E e) {
        
        	// ！！！直接调用offer方法，若失败，就抛异常
            if (offer(e))
                return true;
            else
                throw new IllegalStateException("Queue full");
    }

### enqueue() ###

入队时数组的相关操作：

1.  获取原数组
2.  在 putIndex（队尾） 位置放入元素
3.  计算下一次 putIndex（队尾） ，若已经到数组末尾就将 putIndex 置为队首
4.  随机唤醒一个出队时阻塞在 notEmpty 条件队列的线程

private void enqueue(E x) {
        
            // assert lock.getHoldCount() == 1; 同一时刻只能一个线程进行操作此方法
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            // putIndex 为本次插入的位置
            items[putIndex] = x;
            // 计算下次putIndex位置，若到最后了就将index置为0
        	// 注：保证不会覆盖队首的前提条件是，count==length时在上面三个方法就已经阻塞或返回了
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            // 唤醒取出线程
            notEmpty.signal();
    }

## 2.出队 ##

### take()：队空阻塞 ###

public E take() throws InterruptedException {
        
            final ReentrantLock lock = this.lock; 
            lock.lockInterruptibly(); // 加锁，设置为可中断
            try {
        
                // ！！！队列如果是空的，就将当前线程加入到notEmpty条件队列中，然后进入阻塞状态
                // 等待某个线程拿向队列放入元素（enqueue）后，有机会被唤醒进入同步队列
                // 注：这里的while循环有double-check的意思，即防止线程已经被调度执行了，但前一刻有另一个线程take/poll/remove成功了，所以他又要进入条件队列notFull中阻塞等待。
                while (count == 0)
                    notEmpty.await();
                
                return dequeue(); // 出队时数组的相关操作
            } finally {
        
                lock.unlock(); // 释放锁
            }
        }

### poll()：队空返回null ###

public E poll() {
        
            final ReentrantLock lock = this.lock;
            lock.lock(); // 加锁
            try {
        
                // ！！！队列为空，直接返回null
                return (count == 0) ? null : dequeue(); // 出队时数组的相关操作
            } finally {
        
                lock.unlock(); // 释放锁
            }
    }

### remove()：队空抛异常 ###

思路同上面的add方法，都是直接调用offer/poll，但在 ArrayBlockingQueue 中没有直接实现该方法，而是在它的父类 AbstractQueue 中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

### dequeue() ###

出队时数组的相关操作：

1.  获取原数组
2.  将 takeIndex（队首） 位置的元素删除（置为null）
3.  计算下一次 takeIndex（队首），若已经到达数组末尾就将 takeIndex 置为队首
4.  随机唤醒一个入队时阻塞在 notFull 条件队列的线程

private E dequeue() {
        
            // assert lock.getHoldCount() == 1;
            // assert items[takeIndex] != null;
            final Object[] items = this.items;
            // takeIndex 代表本次拿数据的位置，是上一次拿数据时计算好的
            E x = (E) items[takeIndex];
            // 帮助 gc
            items[takeIndex] = null;
            // ++ takeIndex 计算下次拿数据的位置，若idx=length，idx=0
            if (++takeIndex == items.length)
                takeIndex = 0;
            // 队列实际大小减 1
            count--;
            if (itrs != null)
                itrs.elementDequeued();
            // 唤醒要放入的线程
            notFull.signal();
            return x;
    }

## 3.获取队首：peek ##

public E peek() {
        
            final ReentrantLock lock = this.lock;
        	lock.lock(); // 加锁
            try {
        
                // 循环队列，返回takeIndex位置的元素
                return itemAt(takeIndex); 
            } finally {
        
                lock.unlock(); // 释放锁
            }
    }

## 4.删除 ##

### remove()：删除指定元素 ###

这个方法其实就是寻找要删除元素的索引，然后调用 removeAt 方法执行删除

// 传入要删除的元素，返回删除成功或者失败
    public boolean remove(Object o) {
        
            if (o == null) return false;
            final Object[] items = this.items;
            final ReentrantLock lock = this.lock;
            lock.lock(); // 加锁
            try {
        
            	// 判断队列中是否有元素，若为空则放回false
                if (count > 0) {
        
                    final int putIndex = this.putIndex;
                    int i = takeIndex;
                    // do-while循环寻找要删除元素的索引，然后调用removeAt方法执行删除
                    do {
        
                        if (o.equals(items[i])) {
        
                            removeAt(i);
                            return true;
                        }
                        // ！！重点：如果到数组末尾了，就将i置为0
                        if (++i == items.length)
                            i = 0;
                    } while (i != putIndex);
                }
                return false;
            } finally {
        
                lock.unlock(); // 释放锁
            }
        }

### removeAt()：删除指定位置 ###

删除指定idx的元素，这里分为了两种情况：

1）情况一：removeIndex == takeIdx

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1][]

1.  将 removeIndex 的元素置为null，表示删除了指定元素
2.  takeIdx后移（++），表示下次直接取下一个元素

2）情况二：removeIndex != takeIdx，这里又分为两种情况：

1.  removeIndex+1 == putIdx，直接将putIdx前移一位  
    ![20210222201950208.png_pic_center][]
2.  removeIndex+1 != putIdx，通过for循环将removeIndex与putIdx间元素前移一位，直到removeIndex+1 == putIndex时执行上面情况一  
    ![20200928005537699.png_pic_center][]

void removeAt(final int removeIndex) {
        
        final Object[] items = this.items;
        // 情况一： 删除位置正好等于下次要拿数据的位置
        if (removeIndex == takeIndex) {
        
            // 下次要拿数据的位置直接置空
            items[takeIndex] = null;
            // 要拿数据的位置往后移动一位
            if (++takeIndex == items.length)
                takeIndex = 0;
            // 当前数组的大小减一
            count--;
            if (itrs != null)
                itrs.elementDequeued();
        // 情况二：删除位置不等于下次要拿数据的位置
        } else {
        
            final int putIndex = this.putIndex;
            for (int i = removeIndex;;) {
        
                // 找到要删除元素的下一个
                int next = i + 1;
                if (next == items.length)
                    next = 0;
                // 2.1 下一个元素不是 putIndex
                if (next != putIndex) {
        
                    // 下一个元素往前移动一位
                    items[i] = items[next];
                    i = next;
                // 2.2 下一个元素是 putIndex
                } else {
        
                    // 删除元素
                    items[i] = null;
                    // 下次放元素时，应该从本次删除的元素放
                    this.putIndex = i;
                    break;
                }
            }
            count--;
            if (itrs != null)
                itrs.removedAt(removeIndex);
        }
        notFull.signal();
    }

[JUC]: https://blog.csdn.net/weixin_43935927/article/details/113952868
[JUC_ArrayBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108833582
[JUC_LinkedBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108853656
[JUC_DelayQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858129
[JUC_SynchronousQueue]: https://blog.csdn.net/weixin_43935927/article/details/108866902
[JUC_SynchronousQueue 1]: https://blog.csdn.net/weixin_43935927/article/details/113964294
[JUC_ --]: https://blog.csdn.net/weixin_43935927/article/details/108888294
[JUC 1]: https://blog.csdn.net/weixin_43935927/article/details/108857735
[AQS]: https://blog.csdn.net/weixin_43935927/article/details/108693296
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/bcffa32da4e64d04bdf9094e298fc21f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/d3fa1fd1ba1044d0b683bdd8099ea657.png
[20210222201950208.png_pic_center]: /images/20221123/23d2b121ceba415490af6cf587d1a405.png
[20200928005537699.png_pic_center]: /images/20221123/8b4f481357874c9b9e7e18c6fa59da0b.png