// 基于 JDK 17。链表阻塞队列：takeLock/putLock 两锁分离，头尾各管各的，吞吐高于单锁的 ArrayBlockingQueue。
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    static class Node<E> {
        E item;
        Node<E> next;       // 出队后会指向自己(self-link)，帮 GC 断开跨代引用
        Node(E x) { item = x; }
    }

    private final int capacity;        // 容量上限。默认构造给 Integer.MAX_VALUE，"有界"名存实亡

    private final AtomicInteger count = new AtomicInteger();   // 两把锁唯一共享的状态，必须原子

    transient Node<E> head;            // 不变式：head.item == null，是个哨兵。只有 takeLock 侧动它
    private transient Node<E> last;    // 不变式：last.next == null。只有 putLock 侧动它

    private final ReentrantLock takeLock = new ReentrantLock();    // take/poll 用
    private final Condition notEmpty = takeLock.newCondition();    // 消费者在这上面等"非空"

    private final ReentrantLock putLock = new ReentrantLock();     // put/offer 用
    private final Condition notFull = putLock.newCondition();      // 生产者在这上面等"非满"

    // put 侧发现队列由空变非空时调用：signal 别人锁上的 Condition 必须先持有那把锁
    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();                // 跨锁唤醒的代价：put 线程偶尔要摸一下 takeLock
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

    // take 侧发现队列由满变不满时调用，对称
    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

    private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node;        // 只动 last，不碰 head —— 两锁互不干扰的物理基础
    }

    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC          // 自指：既标记"已出队"，又避免老年代节点拽住新节点
        head = first;                   // 出队 = 把下一个节点变成新哨兵
        E x = first.item;
        first.item = null;
        return x;
    }

    // remove/clear/迭代器修改这类要同时动头尾的操作，才需要两把锁全拿
    void fullyLock() {
        putLock.lock();
        takeLock.lock();
    }

    void fullyUnlock() {
        takeLock.unlock();
        putLock.unlock();
    }

    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);        // 默认"无界"。Executors.newFixedThreadPool 用的就是这个构造
    }

    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);    // 哨兵节点：空队列时 head == last，两把锁也不会摸到同一个字段
    }
    // ...
}

// 基于 JDK 17。java.util.concurrent.LinkedBlockingQueue：put 与 take
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final int c;
        final Node<E> node = new Node<E>(e);    // 节点在锁外分配，缩短持锁时间
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) {   // 持有 putLock 时 count 只会被 take 减小，读到"满"顶多虚惊一场
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();        // c 是加之前的旧值
            if (c + 1 < capacity)               // 加完还没满 -> 顺手唤醒下一个生产者(级联唤醒)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)                             // 旧值为 0 = 队列刚从空变非空，才需要跨锁叫醒消费者
            signalNotEmpty();
    }
    // ...
    public E take() throws InterruptedException {
        final E x;
        final int c;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)                          // 取完还有剩 -> 级联唤醒下一个消费者
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)                      // 旧值为 capacity = 刚从满变不满，才跨锁叫醒生产者
            signalNotFull();
        return x;
    }

为什么不直接用 signalAll? 因为 signalAll 会让全部等待者同时涌起来抢锁, 抢到的只有一个, 其余全部再睡回去, 白白付出 N 次上下文切换; 级联唤醒让每次唤醒都"有位置可用"才发生, 几乎不空转。
后者的条件 `c == 0`(旧值为空)同样是为了省: 跨锁 signal 要先 lock 对面的锁, 是双锁模型里唯一的交叉点, 如果每次 put 都做, 两把锁就退化成一把; 只在"空转非空"这个边沿触发, 是因为只有这一刻才可能有消费者在 notEmpty 上睡——队列非空时来的消费者根本不会睡。