LinkedBlockingQueue的put(E e)与take()方法内部实现解析

Let's break down how the put(E e) and take() methods work in Java's LinkedBlockingQueue—a bounded blocking queue backed by a linked list. These methods are core to its blocking behavior, letting threads wait safely when the queue is empty (for take()) or full (for put()).

先看take()方法的实现逻辑 #

You provided the core code for take(), so let's walk through it step by step:

public E take() throws InterruptedException {
    E x;
    int c = -1;
    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) signalNotFull();
    return x;
}

Here's what each part does:

• 锁获取：takeLock.lockInterruptibly() 获取取操作的独占锁，并且支持中断——如果线程在等待锁的过程中被中断，会直接抛出InterruptedException，比普通的lock()更适配需要响应中断的场景。
• 等待队列非空：while (count.get() == 0) 循环检查队列是否为空。这里用while而非if是为了防范虚假唤醒——线程可能在没有收到合法信号的情况下被唤醒，此时必须重新检查队列状态才能继续。notEmpty.await()会让线程挂起在绑定到takeLock的notEmpty条件上，直到其他线程发出队列有元素的信号。
• 出队操作：x = dequeue() 移除队列头部元素，因为已经持有takeLock，所以这个操作是线程安全的，不会被其他取操作干扰。
• 更新元素计数：c = count.getAndDecrement() 原子性地获取当前队列元素数并递减，c存储的是取元素之前的队列大小。
• 唤醒其他等待的取线程：如果c > 1，说明取走一个元素后队列中至少还有一个元素，此时调用notEmpty.signal()唤醒其他等待获取元素的线程。
• 释放锁：finally块保证无论是否发生异常，都会释放takeLock，避免锁泄漏。
• 唤醒等待的存线程：如果c == capacity，说明取元素之前队列是满的，现在有空位了，调用signalNotFull()唤醒等待添加元素的线程。

再看put(E e)方法的实现逻辑 #

虽然你没提供完整的put()代码，但它的逻辑和take()完全对称，负责将元素添加到队列尾部。核心实现如下：

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock putLock = this.putLock;
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(e);
        c = count.getAndIncrement();
        if (c < capacity) notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0) signalNotEmpty();
}

put()的关键细节：

• 空元素检查：首先会抛出NullPointerException——LinkedBlockingQueue不允许添加null元素。
• 锁获取：调用putLock.lockInterruptibly()获取存操作的独占锁，同样支持中断。
• 等待队列有空位：while (count.get() == capacity) 循环让线程挂起在绑定到putLock的notFull条件上，直到队列有可用空间，同样用while防范虚假唤醒。
• 入队操作：enqueue(e) 将元素添加到队列尾部，因为持有putLock，所以操作安全。
• 更新元素计数：c = count.getAndIncrement() 获取添加元素前的队列大小，然后原子性地递增计数。
• 唤醒其他等待的存线程：如果c < capacity，说明添加元素后队列还有空位，调用notFull.signal()唤醒其他等待添加元素的线程。
• 释放锁：finally块保证putLock一定会被释放。
• 唤醒等待的取线程：如果c == 0，说明添加元素之前队列为空，现在有元素可用了，调用signalNotEmpty()唤醒等待取元素的线程。

核心设计亮点 #

• 分离锁设计：takeLock和putLock的分离是LinkedBlockingQueue并发性能优于ArrayBlockingQueue的关键。后者用单一锁管控所有操作，取和存无法并行；而LinkedBlockingQueue允许取和存操作在队列既不空也不满时同时进行，大大降低了锁竞争，提升了吞吐量。
• 原子计数：AtomicInteger count安全地追踪队列元素数量，即使取和存操作并发执行，也能保证计数更新的原子性，无需额外加锁。
• 条件等待机制：使用Condition对象（notEmpty和notFull）避免了忙等，线程只有在队列状态发生有意义的变化时才会被唤醒，效率更高。

内容的提问来源于stack exchange，提问作者Amit Goel