【集合】ArrayBlockingQueue 源码分析

Bertha 。 2022-12-29 09:24 115阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 概述 #

ArrayBlockingQueue 基于数组实现的有界阻塞队列。

# 二 ArrayBlockingQueue 实例 #

package com.jpeony.concurrent.queue;
    
    import java.util.concurrent.ArrayBlockingQueue;
    
    /**
     * @author yihonglei
     */
    public class ArrayBlockingQueueTest {
        public static void main(String[] args) throws InterruptedException {
            ArrayBlockingQueue<String> arrayQueue = new ArrayBlockingQueue<>(2);
            arrayQueue.offer("one");
    
            System.out.println(arrayQueue.take());
        }
    }

# 三 ArrayBlockingQueue 底层实现 #

主要分析 put()/offer() 入队操作，take()/poll() 出队操作。

## 重要属性 ##

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {
       
        /** 存储队列元素的数组 */        
        final Object[] items;
       
        /** 队头指针，指引 take, poll, peek or remove 操作 */
        int takeIndex;
        
        /** 队尾指针，指引 put, offer, or add 操作 */
        int putIndex;
        
        /** 队列中元素个数 */
        int count;
    
        /** Lock 锁 */
        final ReentrantLock lock;
    
        /** 队列不为空通知条件，唤醒 take() 时没有值被阻塞的线程  */
        private final Condition notEmpty;
    
        /** 队列不满通知条件，唤醒 put() 时队列满了被阻塞的线程 */
        private final Condition notFull;
    
        // ......
    }

## 构造器 ##

创建 ArrayBlockingQueue 必须指定大小，有界阻塞队列。

public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }
    
    /**
     * @param capacity 队列大小，即数组大小
     * @param fair     是否公平锁，true 公平锁，false 非公平锁
     */
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        // 初始化数组
        this.items = new Object[capacity];
        // 创建 Lock
        lock = new ReentrantLock(fair);
        // take 等待通知
        notEmpty = lock.newCondition();
        // put 等待通知
        notFull =  lock.newCondition();
    }

## ArrayBlockingQueue\#put()/offer() ##

**offer()**

非阻塞入队，队列满了，返回放入失败。

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();
        }
    }

入队实现逻辑。

private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        // 队列数据
        final Object[] items = this.items;
        // 入队索引，putIndex，默认从 0 开始
        items[putIndex] = x;
        // putIndex 出队指针，当指向队尾时，重置为 0，重新开始入队
        if (++putIndex == items.length)
            putIndex = 0;
        // 队列实际元素个数
        count++;
        // 唤醒 take() 时因为队列为空被阻塞等待出队的线程，继续从队列取值
        notEmpty.signal();
    }

**put()**

阻塞入队，队列满了，阻塞等待到队列有空时尝试继续放入。

public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 当队列满了后，阻塞入队线程，直到有空位被唤醒时，才继续入队
            while (count == items.length)
                notFull.await();
            // 入队逻辑
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

## ArrayBlockingQueue\#take()/poll() ##

**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;
        // 出队索引，takeIndex 默认从 0 开始
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        // 出队后，对应位置置为 null，help GC
        items[takeIndex] = null;
        // 当takeIndex到队尾时，重置出队的索引为 0，从队头重新开始取值
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // 唤醒 put() 时因为队列满了被阻塞等待入队的线程，继续入队
        notFull.signal();
        return x;
    }

**poll()**

非阻塞获取元素。

public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 有则获取，无责拉倒
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }

单独分析下入队和出队的 **putIndex 和 takeIndex 两个指针**，特别是重置为 0 的地方是一个小技巧。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70][]

队列是“先进先出”的，比如图中。

入队：one、two、three、four、five、six，通过 putIndex 索引，从 0 开始往数组插入元素，当插入到队尾时，即 five 元素时，

putIndex 重置为 0，当插入 six 时，即放在之前 one 的位置，one 的位置被出队了。

出队：one、two、three、four、five、six，通过 takeIndex 索引，从 0 开始出队，一开始是 one, 依次往后类推，当到队尾时，即 five 元素时，

takeIndex 重置为 0，当 takeIndex 重置为 0，获取的就是 six，与 putIndex 入队索引结合，保证队列先进先出的特性。

整个数组就空间循环利用。ArrayBlockingQueue 其他代码逻辑都比较直观，顺序数组或者链式数组 最精华的就是 头和尾 指针。

# 四 总结 #

1、ArrayBlockingQueue 基于数组实现的有界阻塞队列；

2、基于 ReentrantLock 实现加锁和解锁机制，以及 Condition 实现等待通知机制，实现队列的“生产-消费模式”；

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70]: /images/20221120/6e27a8d3bb2d452b98ffe12a3a535788.png