java15源码-ArrayBlockingQueue

一 阻塞队列API

二 类图

三 构造方法

// ArrayBlockingQueue.java
final Object[] items; // 存放数据元素的数组
int putIndex; // put脚标
int count; // 队列中存放的元素数量

/**
     * ReentrantLock
     *     - 保证多生产者多消费者场景下线程安全
     *     - 通过条件队列实现阻塞/通知模式
     */
    final ReentrantLock lock;

/**
     * ReentrantLock锁的条件队列
     * 队列已经空了就无法继续take元素 让线程阻塞在条件队列上
     * 阻塞在上面的线程只有队列不为空才会被唤醒 有机会继续竞争take元素
     */
    @SuppressWarnings("serial")  // Classes implementing Condition may be serializable.
    private final Condition notEmpty;

/** Condition for waiting puts */
    /**
     * ReentrantLock锁的条件队列
     * 队列已经满了就无法继续put元素 让线程阻塞在条件队列上
     * 阻塞在上面的线程只有队列不满才会被唤醒 有机会继续竞争put元素
     */
    @SuppressWarnings("serial")  // Classes implementing Condition may be serializable.
    private final Condition notFull;

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

ArrayBlockingQueue使用的数据结构是数组，通过两个指针移动控制元素入队出队，循环往复

阻塞的实现依赖于ReentrantLock的条件队列

四 API

1 put

// ArrayBlockingQueue.java
public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await(); // 队列已经满了就把put线程阻塞在条件队列上 等待有其他线程take走元素唤醒put线程
            this.enqueue(e); // 元素入队操作
        } finally {
            lock.unlock();
        }
    }

// ArrayBlockingQueue.java
private void enqueue(E e) { // 元素入队操作
        // assert lock.isHeldByCurrentThread();
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[this.putIndex] = e;
        if (++putIndex == this.items.length) putIndex = 0; // 移动put脚标
        this.count++; // 更新元素数量
        this.notEmpty.signal(); // 尝试唤醒曾经因为队列空了阻塞的take线程
    }

2 take

// ArrayBlockingQueue.java
public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await(); // 队列已经空了就把当前take线程阻塞在ReentrantLock的条件队列上 等待其他线程put元素后唤醒take线程
            return this.dequeue(); // 元素出队
        } finally {
            lock.unlock();
        }
    }

// ArrayBlockingQueue.java
private E dequeue() {
        // assert lock.isHeldByCurrentThread();
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E e = (E) items[takeIndex]; // 取到的元素
        items[takeIndex] = null; // help GC
        if (++takeIndex == items.length) takeIndex = 0; // 移动take脚标
        count--; // 元素数量跟新
        if (this.itrs != null)
            this.itrs.elementDequeued();
        this.notFull.signal(); // 尝试唤醒曾经可能因为队列满了而阻塞的put线程
        return e;
    }

五 总结