承接的问题
6.3 的选型矩阵给过一句话：ArrayBlockingQueue 是单锁，LinkedBlockingQueue 是双锁，后者生产者和消费者不互斥。这句话是对的，但它没有回答三个真正影响选型的问题：单锁为什么配两个条件？双锁为什么反而要一个 AtomicInteger？既然双锁不互斥，为什么实测吞吐没有稳定胜出？
这一节只拆这两个类。它们是最常见的“有界缓冲区”，也是理解 Condition 如何把“等空”和“等满”分开的最小样本。AQS 与 Condition 的排队细节第 1、2 章已讲，这里只看队列自己的状态机。
先给结论：
| 维度 | ArrayBlockingQueue | LinkedBlockingQueue |
|---|---|---|
| 锁 | 一把 ReentrantLock | putLock + takeLock |
| 条件 | notEmpty / notFull，同属一把锁 | 各自绑定自己的锁 |
| 元素存储 | 预分配 Object[]，环形下标 | 每元素一个 Node，链表 |
| 计数 | 普通 int，锁保护 | AtomicInteger |
| 默认容量 | 无默认，必须给且大于 0 | Integer.MAX_VALUE |
| 公平模式 | 构造参数可选 | 没有公平构造参数 |
一把锁、两个条件：ArrayBlockingQueue
先看字段。所有可变状态都是普通字段，没有 volatile，因为它们只在 lock 保护下读写：
// JDK 25 源码摘录，已删减
final Object[] items;
int takeIndex;
int putIndex;
int count;
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
transient Itrs itrs;
items 是构造时一次性分配的定长数组，takeIndex 与 putIndex 做环形推进，count 是当前元素数。两个 Condition 由同一把锁创建，所以它们共享同一个 AQS 同步队列，但各自维护一条条件等待队列：等“非空”的消费者进 notEmpty，等“非满”的生产者进 notFull。这就是“一把锁、两个等待集”。
入队与出队的方法体很短：
// JDK 25 源码摘录，已删减
private void enqueue(E e) {
final Object[] items = this.items;
items[putIndex] = e;
if (++putIndex == items.length) putIndex = 0;
count++;
notEmpty.signal();
}
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E e = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length) takeIndex = 0;
count--;
if (itrs != null)
itrs.elementDequeued();
notFull.signal();
return e;
}
注意这里是无条件 signal()：每次入队都唤醒一个等非空的消费者，每次出队都唤醒一个等非满的生产者。它没有先判断“有没有人在等”，因为 Condition.signal() 自己就会检查条件队列是否为空；没有等待者时只是一次廉价判断。更重要的原因是单锁把所有状态变更串行化了：入队时 count 从 0 变 1，此前任何等待中的消费者必然是在 count == 0 时挂起的，所以这一枪不会打空到“语义错误”，最多是“没有等待者”的空枪。
为什么不用 signalAll()？每次变更只让一个等待者具备继续执行的条件：一次入队只增加一个元素，只能满足一个消费者。signalAll() 会把所有等待者搬到同步队列，醒来后只有一个能拿到元素，其余的重新挂起，白白制造一次“惊群”。只有在清空队列、批量删除这类一次改变很多状态的路径上，signalAll() 才是必要的。
公平模式在构造器里直接传给 ReentrantLock：
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();
}
fair = true 时，锁的获取按 FIFO，等待的消费者与生产者不会无限期被后来者插队；代价是每次 lock 都可能要检查队列前驱，吞吐下降。这个参数只影响“谁先拿到锁”，不改变队列的 FIFO 元素顺序。
两把锁、各一个条件：LinkedBlockingQueue
LinkedBlockingQueue 把“改队尾”和“改队头”拆成两条互斥路径：
// JDK 25 源码摘录，已删减
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
private final int capacity;
private final AtomicInteger count = new AtomicInteger();
transient Node<E> head;
private transient Node<E> last;
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
head 是一个 item == null 的哨兵节点，last 指向尾节点；生产者只改 last.next 与 count，消费者只改 head 与 count。两条路径唯一的共享可变状态就是 count。这就是它必须是 AtomicInteger 的原因，而不是“作者顺手用了个原子类”。
第一，count 在一个锁下写、在另一个锁下读。生产者在 putLock 里 getAndIncrement()，消费者在 takeLock 里的等待条件 while (count.get() == 0) 读它。普通 int 没有跨线程可见性保证，AtomicInteger 的读写自带 volatile 语义，把这个跨锁的握手补上。
第二，count 的“读旧值再改”必须原子。生产与消费可以真正并行：生产者执行 count++ 的同时消费者执行 count--，如果用 volatile int，两个读改写会丢更新。更关键的是旧值参与了决策：c == 0 意味着“这次 put 之前队列是空的”，只有这个瞬间才可能有消费者在等；c == capacity 意味着“这次 take 之前队列是满的”，只有这个瞬间才可能有生产者在等。getAndIncrement() 返回的旧值让这个判断没有竞态窗口。
第三，size() 可以无锁：直接 count.get()。这也是为什么不用 LongAdder——LongAdder 擅长“只关心总和”的计数，但它不返回自增前的精确旧值，sum() 也不是一个原子快照，无法支撑上面两个边界判断。
put 的完整路径把这三件事串起来：
// JDK 25 源码摘录，已删减
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 {
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
三个动作的顺序值得逐行看：先在 putLock 下等到有空位，再把节点挂到尾部，然后原子自增拿到旧值 c。如果 c + 1 < capacity，说明入队后还有空位，可以再唤醒一个等满的生产者；如果入队后正好满了，唤醒也没用，因为被唤醒者会重新检查 count == capacity 再挂起。最后，只有 c == 0（入队前为空）才去唤醒消费者，而且是在释放 putLock 之后通过 signalNotEmpty() 去拿 takeLock。
这个“释放后再拿另一把锁”的顺序不是风格问题。若生产者在持有 putLock 时去拿 takeLock，消费者又在持有 takeLock 时去拿 putLock，就会形成锁顺序反转的死锁。JDK 的写法保证任何线程最多持有一把队列锁；跨锁唤醒一定是“先放自己的，再拿对方的”。
take 是对称的：
// JDK 25 源码摘录，已删减
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)
signalNotFull();
return x;
}
c > 1 表示取走一个后还有元素，可以再唤醒一个消费者；c == capacity 表示取走前是满的，取走后腾出了一个位置，必须唤醒一个生产者。两个判断都基于“等待者只可能出现在边界上”这一不变式。
只在必要时唤醒：signal 的判据
把两个类的唤醒条件并排放，差异就清楚了：
| 操作 | ArrayBlockingQueue | LinkedBlockingQueue |
|---|---|---|
| 入队后 | 无条件 notEmpty.signal() | 仅 c == 0 时 signalNotEmpty() |
| 入队后仍有空位 | 不需要额外动作 | c + 1 < capacity 时 notFull.signal() |
| 出队后 | 无条件 notFull.signal() | 仅 c == capacity 时 signalNotFull() |
| 出队后仍有元素 | 不需要额外动作 | c > 1 时 notEmpty.signal() |
ArrayBlockingQueue 的“无条件”并不表示它更笨。单锁下，入队与出队本身已经互斥，signal() 的成本只是检查条件队列头指针；它省掉的是“判断有没有等待者”的分支，而不是唤醒。真正的唤醒只发生在确实有等待者时。
LinkedBlockingQueue 则不能无条件：它的 notEmpty 属于 takeLock，生产者要唤醒消费者必须去拿另一把锁。如果每次 put 都拿一次 takeLock，双锁就退化成了“每次操作两把锁”，拆分锁的收益会被这笔额外获取吃掉。于是它用 count 的旧值做精确判定，只在队列“从空变非空”或“从满变非满”的瞬间跨锁。这是双锁设计里最容易被忽略的一笔账：锁拆开之后，跨锁通信必须变成稀有事件，否则拆分没有意义。
容量语义与这个设计的代价
ArrayBlockingQueue 的容量是构造时固定的数组长度，创建即分配 capacity 个引用槽；LinkedBlockingQueue 的容量是上界，节点按需分配，默认 Integer.MAX_VALUE。这带来两个完全不同的失败模式：
| 场景 | ArrayBlockingQueue | LinkedBlockingQueue |
|---|---|---|
| 容量给大 | 构造时立刻占用内存 | 空队列几乎不占内存，涨到容量才封顶 |
| 容量给成默认 | 不存在默认，必须显式给 | 等于无界，生产快于消费时涨到 OOM |
| 满了之后 | put 阻塞，offer 返回 false | 同样，但有界必须显式设置 |
| 公平性 | 可选 fair | 不支持 |
三个 API 的语义也要分清：put 满时阻塞且响应中断；offer(e, timeout, unit) 超时返回 false；offer(e) 立即返回 false；add(e) 满时抛 IllegalStateException。remainingCapacity() 只是一个瞬时估计，返回值在下一纳秒就可能过期，不能拿它做“一定能放进去”的判断——唯一权威的判断是 offer 的返回值。
这个设计的代价也很明确：
ArrayBlockingQueue的入队与出队共用一把锁。即使队列里既有空位又有元素，生产者和消费者也不能同时修改结构；临界区里还包含signal()和迭代器维护，临界区越长，锁队列越长。LinkedBlockingQueue虽然拆了锁，但count是两边每操作一次就要碰的共享原子变量，head、last与它在对象里相邻，缓存行仍然会在核心之间来回迁移；拆锁消掉的是互斥，不是全部竞争。7.3 会用 JOL 的字段偏移展开这一点。LinkedBlockingQueue没有公平构造参数，两个ReentrantLock都是非公平的。如果业务真的需要“等待者按到达顺序被服务”，只能选ArrayBlockingQueue(capacity, true)，或者自己在外面加一层公平调度。
什么时候不要用它们：需要零容量直接交接时用 SynchronousQueue；需要按优先级或到期时间出队时用 7.2 的两个类；需要“生产者和消费者都不阻塞、只是传递引用”时用 ConcurrentLinkedQueue。这两个类是“有界缓冲 + 背压”的工具，不是通用消息总线。
实测：唤醒次数与吞吐
先看唤醒次数。为了不依赖对 JDK 私有字段的反射，我写了一个同构最小实现，只保留两个类的锁、条件与 signal 路径，并在每次 signal 后计数；生产/消费总量固定为 100 万，用 CountDownLatch 对齐起跑线。本机输出（负载均值 110.5，10 核）：
capacity=1024, loadavg=110.5458984375
ABQ P=1 C=1 total=1,000,000 time=0.113s ops/s=17,645,833 notEmpty.signal=1,000,000 (1.000/put) notFull.signal=1,000,000 (1.000/take)
LBQ P=1 C=1 total=1,000,000 time=0.157s ops/s=12,698,745 notEmpty.signal=969 (0.001/put) notFull.signal=46 (0.000/take)
ABQ P=4 C=4 total=1,000,000 time=0.063s ops/s=31,677,356 notEmpty.signal=1,000,000 (1.000/put) notFull.signal=1,000,000 (1.000/take)
LBQ P=4 C=4 total=1,000,000 time=0.082s ops/s=24,515,662 notEmpty.signal=945 (0.001/put) notFull.signal=152 (0.000/take)
ABQ P=8 C=8 total=1,000,000 time=0.108s ops/s=18,442,614 notEmpty.signal=1,000,000 (1.000/put) notFull.signal=1,000,000 (1.000/take)
LBQ P=8 C=8 total=1,000,000 time=0.088s ops/s=22,726,885 notEmpty.signal=731 (0.001/put) notFull.signal=418 (0.000/take)
这组数字要分两半看。唤醒次数是可信的：ABQ 每次 put 都调用一次 notEmpty.signal()，每次 take 都调用一次 notFull.signal()；LBQ 在消费者跟得上的工作负载里，队列几乎不回到空，生产者很少看到 c == 0，所以 100 万次 put 只触发了不到 1000 次跨锁唤醒。这正是“只在必要时唤醒”的直观效果。
吞吐列不可信，因为探针在每次 signal 里做了一次 AtomicLong.incrementAndGet()，ABQ 被计了 200 万次，LBQ 只有约 2000 次，这个开销把 ABQ 的 ops/s 压低了。它只能说明“同构实现能跑起来”，不能拿来比较两个类的吞吐。
吞吐用 JMH 1.37 单独测，@Group 一组生产者一组消费者，容量 1024，2 次 1 秒预热、3 次 1 秒测量、1 个 fork。运行前的负载均值是 65.48 / 85.44 / 96.38（1/5/15 分钟），结果如下：
# 本机 JMH 输出摘录：四个 Result 段各自保留原始数值与格式
# Parameters: (kind = abq)
Result "jmh.QueueBench1.four":
19760078.719 ±(99.9%) 33319033.370 ops/s [Average]
# Parameters: (kind = lbq)
Result "jmh.QueueBench1.four":
29000335.252 ±(99.9%) 19947152.296 ops/s [Average]
# Parameters: (kind = abq)
Result "jmh.QueueBench1.one":
18833118.686 ±(99.9%) 125420927.951 ops/s [Average]
# Parameters: (kind = lbq)
Result "jmh.QueueBench1.one":
11399168.976 ±(99.9%) 41186667.335 ops/s [Average]
单生产者单消费者时 ABQ 的均值更高，4 对 4 时 LBQ 更高，方向随线程数翻转；更重要的是误差项大得离谱，one/abq 的 99.9% 置信区间下界已经是 0，four/lbq 的区间也和 four/abq 大幅重叠。在这台负载常年 60–110 的机器上，两组数字的差异不构成结论，只能确认它们都在 10^7 ops/s 量级。复现命令：
export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
export PATH="$JAVA_HOME/bin:$PATH"
JH="$HOME/.m2/repository/org/openjdk/jmh"
javac --release 25 -d out -cp "$JH/jmh-core/1.37/jmh-core-1.37.jar" \
-processorpath "$JH/jmh-generator-annprocess/1.37/jmh-generator-annprocess-1.37.jar:$JH/jmh-core/1.37/jmh-core-1.37.jar" \
bench/QueueBench1.java
java -cp "out:$JH/jmh-core/1.37/jmh-core-1.37.jar:\
$HOME/.m2/repository/net/sf/jopt-simple/jopt-simple/5.0.4/jopt-simple-5.0.4.jar:\
$HOME/.m2/repository/org/apache/commons/commons-math3/3.6.1/commons-math3-3.6.1.jar" \
jmh.QueueBench1 -wi 2 -i 3 -w 1s -r 1s -f 1
在 libraryd-conc 里落地
libraryd-conc 的借还事件要送给通知扇出（mail、sms、push）。这是一个典型的有界缓冲：峰值每秒约 2000 个事件，下游最慢时允许积压 2 秒，于是容量取 4096（2000 × 2 向上取 2 的幂）。这里选 ArrayBlockingQueue 而不是 LBQ，理由不是吞吐，而是三件事：容量固定、内存可预测、需要公平模式避免还书事件被借书事件长期插队。
package com.plume.conc.core;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
/** 借还事件的有界缓冲：满了就让调用方拿到 false，自己做降级。 */
public final class LoanEventBus {
public record LoanEvent(String isbn, String memberId, Type type) {
public enum Type { BORROW, RETURN, RESERVE }
}
private final BlockingQueue<LoanEvent> queue;
public LoanEventBus(int capacity) {
this.queue = new ArrayBlockingQueue<>(capacity, true);
}
/** 返回 false 表示缓冲已满，调用方应记录并降级，而不是无限等待。 */
public boolean publish(LoanEvent event, long timeoutMillis) throws InterruptedException {
return queue.offer(event, timeoutMillis, TimeUnit.MILLISECONDS);
}
public LoanEvent take() throws InterruptedException {
return queue.take();
}
/** 批量取，减少下游每次只处理一条的调度开销。 */
public List<LoanEvent> drain(int max) {
List<LoanEvent> batch = new ArrayList<>(max);
queue.drainTo(batch, max);
return batch;
}
public int backlog() {
return queue.size();
}
}
调用方必须处理 publish 的 false：这正是有界队列的价值——把“下游扛不住”变成调用点上一个可观测、可决策的返回值，而不是几小时后堆里堆满 Node 的 OOM。如果事件处理是纯内存、且生产与消费都很猛、临界区里还要做序列化，那么 LBQ 的双锁更合适；但容量仍要显式给，不能吃默认值。
package com.plume.conc.core;
public final class LoanEventBusDemo {
public static void main(String[] args) throws Exception {
LoanEventBus bus = new LoanEventBus(2);
Thread consumer = Thread.ofPlatform().start(() -> {
try {
System.out.println("consumed " + bus.take());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
System.out.println(bus.publish(
new LoanEventBus.LoanEvent("978-7-111-1", "m-1",
LoanEventBus.LoanEvent.Type.BORROW), 100));
System.out.println(bus.publish(
new LoanEventBus.LoanEvent("978-7-111-2", "m-2",
LoanEventBus.LoanEvent.Type.RETURN), 100));
System.out.println("backlog=" + bus.backlog());
consumer.join();
}
}
常见错误与边界
| 错误 | 后果 | 正确做法 |
|---|---|---|
new LinkedBlockingQueue<>() | 默认容量 Integer.MAX_VALUE，等于无界，最终 OOM | 显式给容量，或用 ArrayBlockingQueue |
用 remainingCapacity() 判断能否放入 | 判断与 put 之间状态已变 | 直接用 offer 的返回值 |
以为 size() 是稳定快照 | 并发下它只是某一瞬间的读数 | 监控趋势，不做业务判断 |
在 ABQ 上开 fair = true 还期待高吞吐 | 公平锁每次都要检查前驱 | 只有确实需要防饥饿时才开 |
| 在 LBQ 上找公平构造参数 | 不存在，两个锁都是非公平 | 要公平只能换 ABQ，或外置调度 |
自己扩展队列时先拿 putLock 再拿 takeLock | 与 JDK 的加锁顺序相反，可能死锁 | 任何时刻只持有一把锁 |
还有一个不常被提到的边界：ArrayBlockingQueue 的 count、takeIndex、putIndex 都是普通 int，迭代器 Itrs 也在锁下维护。这意味着它的 iterator() 不是快照，而是弱一致视图；遍历时队列被其他线程修改，不抛 ConcurrentModificationException，但可能看到部分更新。需要稳定快照时，应该在锁外自己复制，或者换用 6.1 的写时复制结构。
小结
ArrayBlockingQueue 用一把锁把数组、下标、计数和两个等待集串起来，简单、可公平、内存可预测；LinkedBlockingQueue 用两把锁把队头与队尾拆开，代价是必须用 AtomicInteger 维护跨锁的 count，并且只在“空→非空”“满→非满”的边界上做跨锁唤醒。前者适合容量小、要公平、要确定内存的场景，后者适合生产消费都猛、能接受每元素一个 Node 的场景。两者的吞吐差异在本机高负载下没有稳定方向，选型应先看容量语义与失败模式，再看锁的拓扑。
7.2 转向三个“不存储”或“非 FIFO”的队列：零容量的 SynchronousQueue、按堆序出队的 PriorityBlockingQueue、按到期时间出队的 DelayQueue。
阅读导航：上一节 6.3 并发集合选型矩阵 ｜ 下一节 7.2 SynchronousQueue、Priority、Delay
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。