// 基于 JDK 17 (本地 JAVA_Source 仓), java.util.concurrent.ArrayBlockingQueue
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    final Object[] items;               // 定长数组, 创建后永不扩容, "有界"由此而来

    int takeIndex;                      // 下一个 take/poll/peek 的位置(队头)

    int putIndex;                       // 下一个 put/offer 的位置(队尾), 与 takeIndex 构成环形缓冲

    int count;                          // 不是 volatile, 可见性全靠 lock, 所以连 size() 都要加锁

    final ReentrantLock lock;           // 一把锁管全部读写。与 LinkedBlockingQueue 双锁的关键分野

    private final Condition notEmpty;   // take 阻塞在这, enqueue 后 signal

    private final Condition notFull;    // put 阻塞在这, dequeue 后 signal

    transient Itrs itrs;                // 追踪所有活跃迭代器, 出队/删除时同步修正它们(弱一致迭代的代价)

    // ...

    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);         // fair 只决定锁的排队公平, 元素本身永远 FIFO
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    // 只在持锁时调用, 所以内部全是普通读写, 一个 CAS 都没有
    private void enqueue(E e) {
        // assert lock.isHeldByCurrentThread();
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = e;
        if (++putIndex == items.length) putIndex = 0;  // 到尾绕回 0: 环形数组, 出入队都 O(1) 免搬移
        count++;
        notEmpty.signal();      // 只多了一个元素, 精准叫醒一个 take 即可, signalAll 是纯惊群
    }

    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;    // 断引用, 否则出队元素被数组钉住无法 GC
        if (++takeIndex == items.length) takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();     // 队头动了, 通知迭代器组检查自己是否已"过期"
        notFull.signal();
        return e;
    }
}

// 基于 JDK 17 (本地 JAVA_Source 仓), java.util.concurrent.ArrayBlockingQueue
    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);      // BlockingQueue 禁 null: poll 返回 null 表示"空", 二义性不允许
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();       // 等锁期间可被中断, 阻塞方法的标配
        try {
            while (count == items.length)   // while 防虚假唤醒: 从 signal 到重新持锁有窗口, 条件可能又被破坏
                notFull.await();            // 原子地"释放锁+挂起", 醒来时已重新持锁
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    public boolean offer(E e) {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;       // 满了立刻返回 false 不阻塞: 线程池靠它感知"队列满", 才会去开非核心线程
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {

        Objects.requireNonNull(e);
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length) {
                if (nanos <= 0L)
                    return false;
                nanos = notFull.awaitNanos(nanos);  // 返回剩余等待时间: 被提前唤醒后接着等余额, 总时长不超 timeout
            }
            enqueue(e);
            return true;
        } finally {
            lock.unlock();
        }
    }

// 基于 JDK 17 (本地 JAVA_Source 仓), java.util.concurrent.ArrayBlockingQueue
    void removeAt(final int removeIndex) {
        // ...
        final Object[] items = this.items;
        if (removeIndex == takeIndex) {
            // removing front item; just advance
            items[takeIndex] = null;    // 删队头是特例: 等价一次 dequeue, O(1)
            if (++takeIndex == items.length) takeIndex = 0;
            count--;
            if (itrs != null)
                itrs.elementDequeued();
        } else {
            // an "interior" remove

            // slide over all others up through putIndex.
            for (int i = removeIndex, putIndex = this.putIndex;;) {     // 从删除点到队尾逐个前移, O(n) 且全程持锁
                int pred = i;
                if (++i == items.length) i = 0;
                if (i == putIndex) {
                    items[pred] = null;
                    this.putIndex = pred;
                    break;
                }
                items[pred] = items[i];
            }
            count--;
            if (itrs != null)
                itrs.removedAt(removeIndex);    // 中间删除让所有迭代器手里的下标失真, 必须集中修正
        }
        notFull.signal();
    }

为什么条件检查必须 `while` 不能 `if`:`signal` 只是把节点挪进同步队列,从被唤醒到真正拿回锁之间有窗口,另一个 `offer` 可能插队把队列重新填满;加上 `park` 本身允许虚假唤醒,所以醒来后条件必须重查。
为什么要两个 Condition:`synchronized` 只有一个 wait 集合,`notify` 可能叫醒同类线程(put 叫醒 put),只能 `notifyAll` 惊群兜底;两个条件队列把生产者和消费者物理分开,put 永远只叫醒 take,`signal` 单个就够。
为什么坚持单锁:`count`、`putIndex`、`takeIndex` 被出入两端共享读写,拆成两把锁就得像 `LinkedBlockingQueue` 那样把 count 换成 `AtomicInteger` 并小心处理跨锁的级联 signal;数组两端还可能物理上指向同一个槽位。