基于AQS的ArrayBlockingQueue源码实现

男娘i 2024-04-17 06:00 15阅读 0赞

## 基于AQS的ArrayBlockingQueue源码实现 ##

### 一、引言 ###

#### 1.1 普通队列 ####

譬如我们经常见到的`ArrayList`、`LinkedList`的队列，这些可以称之为简单队列或者普通队列。对它们的使用和理解我们就不做过多介绍。

> 说明 本人研究的JDK版本为JDK8。不同版本的实现可能略有差异，请读者们周知。

#### 1.2 阻塞队列 ####

`JDK5`后，Java在`J.U.C`包下给我们提供了一系列的`阻塞队列`的实现，譬如常见的`ArrayBlockingQueue`、`LinkedBlockQueue`、`DelayQueue`等七个不同功能实现的阻塞队列。

但是它和普通队列有什么区别呢，主要用来做什么？值得我们去思考和探究。

**不同点**

阻塞队列支持阻塞添加和阻塞删除方法。

*  **阻塞添加** 所谓的阻塞添加是指当阻塞队列元素已满时，队列会阻塞加入元素的线程，直队列元素不满时才重新唤醒线程执行元素加入操作。
 *  **阻塞删除** 阻塞删除是指在队列元素为空时，删除队列元素的线程将被阻塞，直到队列不为空再执行删除操作(一般都会返回被删除的元素)

**应用场景**

生产者和消费者是我们在实际开发中经常遇到的场景，在没有出现阻塞队列之前，如果我们想要利用队列要生产者和消费者的平衡，必须我们手动使用线程的方法来实现其二者的平衡，但是，当阻塞队列出现之后，我们就可以摈弃这些细节，更加专注于我们自己的业务逻辑的处理。

**示例代码**

下面演示了两个消费者和一个生产者之间保持平衡的实例

public class ArrayBlockingQueue01 { 
    
        public static void main(String[] args) { 
            final BlockingQueue<Integer> bq = new ArrayBlockingQueue(10);
            Runnable produce01 = new Runnable(){ 
                int i = 0;
                @Override
                public void run() { 
                    for (;;) { 
                        try { 
                            System.out.println("生产者01生产了一个：" + i);
                            bq.put(i);
                            i++;
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            };
            Runnable customer01 = new Runnable() { 
                @Override
                public void run() { 
                    for (;;) { 
                        try { 
                            System.out.println("消费者01消费了一个：" + bq.take());
                            Thread.sleep(600);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            };
    
            Runnable customer02 = new Runnable() { 
                @Override
                public void run() { 
                    for (;;) { 
                        try { 
                            System.out.println("消费者02消费了一个：" + bq.take());
                            Thread.sleep(500);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                }
            };
            Thread t01 = new Thread(customer01);
            Thread t02 = new Thread(customer02);
            Thread t03 = new Thread(produce01);
            t01.start();
            t02.start();
            t03.start();
        }
    }

**打印结果**

生产者01生产了一个：0
    生产者01生产了一个：1
    生产者01生产了一个：2
    消费者02消费了一个：0
    消费者01消费了一个：1
    生产者01生产了一个：3
    生产者01生产了一个：4
    生产者01生产了一个：5
    生产者01生产了一个：6
    生产者01生产了一个：7
    生产者01生产了一个：8
    生产者01生产了一个：9
    生产者01生产了一个：10
    生产者01生产了一个：11
    生产者01生产了一个：12
    消费者02消费了一个：2
    生产者01生产了一个：13
    消费者01消费了一个：3
    生产者01生产了一个：14
    消费者02消费了一个：4
    生产者01生产了一个：15
    消费者01消费了一个：5
    生产者01生产了一个：16
    消费者02消费了一个：6
    生产者01生产了一个：17
    消费者01消费了一个：7
    生产者01生产了一个：18
    消费者02消费了一个：8
    生产者01生产了一个：19

由上述结果我们可以看出，其二者之间保持了平衡。

### 二、原理 ###

#### 2.1 类结构 ####

**继承关系**

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { }

![在这里插入图片描述][20190825175424564.png]

ArrayBlockingQueue继承自`AbstractQueue` \-> `AbstractCollection` \-> `Collection` \-> `Iterable`。实现了`BlockingQueue` \-> `Queue` \-> `Collection` \-> `Iterable`。是集合和队列的一种综合实现。

**字段**

/** * Serialization ID. This class relies on default serialization * even for the items array, which is default-serialized, even if * it is empty. Otherwise it could not be declared final, which is * necessary here. */
        private static final long serialVersionUID = -817911632652898426L;
    
        /** The queued items */
        // 一个定长数组，维护ArrayBlockingQueue的元素
        final Object[] items;
    
        /** items index for next take, poll, peek or remove */
        // 下一个获取，轮询，查看或删除的项目索引
        int takeIndex;
    
        /** items index for next put, offer, or add */
        // 下一个put，offer或add的items索引
        int putIndex;
    
        /** Number of elements in the queue */
        // 队列中的元素数量
        int count;
    
        /* * Concurrency control uses the classic two-condition algorithm * found in any textbook. */
    
        /** Main lock guarding all access */
        // 主锁保护所有访问
        final ReentrantLock lock;
    
        /** Condition for waiting takes */
        // take等待的条件
        private final Condition notEmpty;
    
        /** Condition for waiting puts */
        // 等待put的条件
        private final Condition notFull;
    
        /** * Shared state for currently active iterators, or null if there * are known not to be any. Allows queue operations to update * iterator state. */
         // 当前活动迭代器的共享状态，如果已知不存在，则返回null。允许队列操作更新迭代器状态
        transient Itrs itrs = null;

**构造方法**

/** * Creates an {@code ArrayBlockingQueue} with the given (fixed) * capacity and default access policy. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity < 1} */
        public ArrayBlockingQueue(int capacity) { 
        	// 看下面，默认是非公平锁
            this(capacity, false);
        }
    
        /** * Creates an {@code ArrayBlockingQueue} with the given (fixed) * capacity and the specified access policy. * * @param capacity the capacity of this queue * @param fair if {@code true} then queue accesses for threads blocked * on insertion or removal, are processed in FIFO order; * if {@code false} the access order is unspecified. * @throws IllegalArgumentException if {@code capacity < 1} */
        public ArrayBlockingQueue(int capacity, boolean fair) { 
        	// 初始化items，lock ，notEmpty ，notFull 
            if (capacity <= 0)
                throw new IllegalArgumentException();
            this.items = new Object[capacity];
            lock = new ReentrantLock(fair);
            notEmpty = lock.newCondition();
            notFull =  lock.newCondition();
        }
    
        /** * Creates an {@code ArrayBlockingQueue} with the given (fixed) * capacity, the specified access policy and initially containing the * elements of the given collection, * added in traversal order of the collection's iterator. * * @param capacity the capacity of this queue * @param fair if {@code true} then queue accesses for threads blocked * on insertion or removal, are processed in FIFO order; * if {@code false} the access order is unspecified. * @param c the collection of elements to initially contain * @throws IllegalArgumentException if {@code capacity} is less than * {@code c.size()}, or less than 1. * @throws NullPointerException if the specified collection or any * of its elements are null */
        public ArrayBlockingQueue(int capacity, boolean fair,
                                  Collection<? extends E> c) { 
            // 看上面说明
            this(capacity, fair);
    		
    		// 加锁，把集合中的元素，一一添加到队列中去，并同时更新putIndex的值为集合的数量大小，takeIndex默认为0，处在队首。
            final ReentrantLock lock = this.lock;
            lock.lock(); // Lock only for visibility, not mutual exclusion
            try { 
                int i = 0;
                try { 
                    for (E e : c) { 
                        checkNotNull(e);
                        items[i++] = e;
                    }
                } catch (ArrayIndexOutOfBoundsException ex) { 
                    throw new IllegalArgumentException();
                }
                count = i;
                putIndex = (i == capacity) ? 0 : i;
            } finally { 
                lock.unlock();
            }
        }

#### 2.2 原理示意图 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTcyMzU0NA_size_16_color_FFFFFF_t_70]

其实我们可以看出，`ArrayBlockingQueue`内部使用`可重入锁ReentrantLoc` \+ `Condition`来完成多线程环境的并发操作，并保证队列的`阻塞`删除和添加。

### 三、源码分析 ###

#### 3.1 入队 ####

**add(E e)**

将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量），在成功时返回 true，如果此队列已满，则抛出 IllegalStateException

**offer(E e)**

将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量），在成功时返回 true，如果此队列已满，则返回 false

**put(E e)**

将指定的元素插入此队列的尾部，如果该队列已满，则等待可用的空间后才进行插入。这个方法也就是我们要重点理解的方法。也即是`阻塞插入`的方法。

> 多说一句：其实这三者的方法的功能是一样，只是对待当队列满时，返回的处理逻辑不同而已，三者之间并没有任何高低优劣之分，还是要根据我们具体的情况来决定使用哪种方法的插入。

**源码探析**

/** * Inserts the specified element at the tail of this queue, waiting * for space to become available if the queue is full. * * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */
        public void put(E e) throws InterruptedException { 
        	// 1. NPE校验
            checkNotNull(e);
            // 2. 获取锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
            	// 3.1 如果发现队列已满，进行await(),释放锁，进入等待队列
                while (count == items.length)
                    notFull.await();
                // 3.2 如果3.1不成立的，则会执行执行下面的enqueue
                // enqueue(e)中主要做的就是入队操作
                enqueue(e);
            } finally { 
                lock.unlock();
            }
        }
    
        /** * Throws NullPointerException if argument is null. * * @param v the element */
        private static void checkNotNull(Object v) { 
            if (v == null)
                throw new NullPointerException();
        }
        
        /** * Inserts element at current put position, advances, and signals. * Call only when holding lock. * 入队操作，加入队列的尾部，同时记得执行signal()的操作，唤醒等待的take操作，叫他起来，获取元素 */
        private void enqueue(E x) { 
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            notEmpty.signal();
        }

#### 3.2 出队 ####

**poll()**

获取并移除此队列的头，如果此队列为空，则返回 null

**remove(Object o)**

从此队列中移除指定元素的单个实例（如果存在）

**take()**

获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。此方法也就是我们常说的`阻塞获取`。

关于这三者的关系就不在多说，和上面`插入`的类似。这里我们重点探讨`阻塞获取`方法的实现。

**源码实现**

public E take() throws InterruptedException { 
        	// 1. 获取锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
            	// 2.1 如果发现队列为空，则调用notEmpty的等待方法，释放锁并进入等待状态。否则就进入2.2步骤中
                while (count == 0)
                    notEmpty.await();
                // 2.2 走到这有两种情况：
                // 其一，获取时，队列中有元素
                // 其二，获取时，队列为空，进入的等待状态。在等待过程中队列中添加了元素，导致队列不为空，所以在上一步唤醒后，继续走到这里。
                // 3. 执行出队操作
                return dequeue();
            } finally { 
                lock.unlock();
            }
        }
    
        /** * Extracts element at current take position, advances, and signals. * Call only when holding lock. * 出队的同时，记得要唤醒不满notFull的等待队列。因为刚刚取了的一个元素，队列不可能是满的。 */
        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;
        }

#### 3.3 其他方法 ####

**element()**

获取但不移除此队列的头元素，没有元素则抛异常

**peek()**

获取但不移除此队列的头；若队列为空，则返回 null

> 可以把上述两种方法理解成一种检查方法。

### 四、总结 ###

#### 4.1 总结 ####

总的来说，`ArrayBlockingQueue`内部确实是通过`数组`对象`items`来存储所有的数据，值得注意的是`ArrayBlockingQueue`通过一个ReentrantLock来同时控制添加线程与移除线程的并非访问，这点与`LinkedBlockingQueue`区别很大(稍后会分析)。而对于`notEmpty`条件对象则是用于`存放等待`或`唤醒`调用`take`方法的线程，**告诉他们队列已有元素，可以执行获取操作**。同理`notFull`条件对象是用于`等待`或`唤醒`调用`put`方法的线程，告诉它们，**队列未满，可以执行添加元素的操作**。

其中takeIndex代表的是下一个方法(take，poll，peek，remove)被调用时获取数组元素的索引，putIndex则代表下一个方法（put, offer, or add）被调用时元素添加到数组中的索引。

#### 4.2 开发注意 ####

实际开发中往往使用`ArrayBlockingQueue`多一些，首先分配固定大小的容量，极大的降低了内存溢出或JVM频繁GC的现象，其次我们业务往往对整个阻塞队列处理任务做一些监控，而`ArrayBlockingQueue`也方便很多。

而使用`LinkedBlockingQueue`则没有上述优势，请读者们注意。

[20190825175424564.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/6b0f292c261147f98750da402715cc8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTcyMzU0NA_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/16/0389539624af412cbe1798e61fe72fb3.png