承接的问题
7.1 的两个队列解决的是“存下来再取走”。但有三类需求不适合存储：生产者必须把元素直接交到某个消费者手里，中间不能有缓冲；元素要按优先级而不是到达顺序出队；元素要等到某个时刻才能被取走。JUC 分别给了 SynchronousQueue、PriorityBlockingQueue、DelayQueue。
这三个类的源码结构差异很大，但都围绕同一个问题：当“能不能取”不由元素数量决定时，等待条件该怎么表达。7.1 的条件是“非空/非满”，这一节的条件分别是“有配对的交接者”“堆顶存在”“堆顶到期”。
先给定位：
| 类 | 容量 | 出队顺序 | 阻塞点 |
|---|---|---|---|
SynchronousQueue | 0 | 由配对线程决定 | 没有配对的生产者/消费者 |
PriorityBlockingQueue | 无界 | 堆序（最小/比较器） | 队列为空 |
DelayQueue | 无界 | 最早到期者 | 堆顶未到期 |
SynchronousQueue：零容量的直接交接
先纠正一个旧版教材里常见的说法。很多资料会写“SynchronousQueue 在非公平模式下用 TransferStack，公平模式下用 TransferQueue”。这句话描述的是 JDK 8 的实现，JDK 25 已经不是这样了；本机 JDK 25 的 SynchronousQueue.java 里 TransferStack 出现 0 次。本机 javap -p 的输出是：
final class java.util.concurrent.SynchronousQueue$Transferer<E> extends java.util.concurrent.LinkedTransferQueue<E> {
java.util.concurrent.SynchronousQueue$Transferer();
final java.lang.Object xferLifo(java.lang.Object, long);
private void unspliceLifo(java.util.concurrent.LinkedTransferQueue$DualNode);
}
Transferer 不再是一个抽象基类，而是直接继承 LinkedTransferQueue，只额外加了 LIFO 模式的 xferLifo。SynchronousQueue 自己的 xfer 只做一次分派：
// JDK 25 源码摘录，已删减
private Object xfer(Object e, long nanos) {
Transferer<E> x = transferer;
return (fair) ? x.xfer(e, nanos) : x.xferLifo(e, nanos);
}
也就是说，公平模式走的是父类 LinkedTransferQueue.xfer 的 FIFO 队列，非公平模式走子类 xferLifo 的 LIFO 栈。两个模式共用同一套 DualNode 与 CAS 链接逻辑，不再维护两套节点类。
xferLifo 的核心循环值得读一遍：
// JDK 25 源码摘录，已删减
static final class Transferer<E> extends LinkedTransferQueue<E> {
final Object xferLifo(Object e, long ns) {
boolean haveData = (e != null);
Object m;
outer: for (DualNode s = null, p = head;;) {
while (p != null) {
boolean isData; DualNode n, u;
if ((isData = p.isData) != ((m = p.item) != null))
p = (p == (u = cmpExHead(p, (n = p.next)))) ? n : u;
else if (isData == haveData) // same mode; push below
break;
else if (p.cmpExItem(m, e) != m)
p = head; // missed; restart
else { // matched complementary node
Thread w = p.waiter;
cmpExHead(p, p.next);
LockSupport.unpark(w);
break outer;
}
}
if (ns == 0L) { m = e; break; }
if (s == null) s = new DualNode(e, haveData);
s.next = p;
if (p == (p = cmpExHead(p, s))) {
if ((m = s.await(e, ns, this, p == null || p.waiter == null)) == e)
unspliceLifo(s);
else if (m != null) s.selfLinkItem();
break;
}
}
return m;
}
}
DualNode 的字段是 volatile Object item、DualNode next、Thread waiter、final boolean isData。isData 区分节点是“送数据”还是“要数据”，item 在匹配时被 CAS 替换。循环先跳过已匹配/取消的栈顶；遇到同模式就准备压栈；遇到互补模式就 cmpExItem(m, e) 把对方的 item 换成自己的值，CAS 成功后弹栈并 unpark 对方线程。非公平模式的“栈”体现在压栈位置：cmpExHead(p, s) 把新节点放到 head，后到的等待者先被匹配。公平模式的父类 xfer 从 head 沿 next 找，先到的先匹配。两者的差别只是链接方向和匹配顺序，底层都是“先尝试无锁匹配，匹配不到才分配节点并 park”。
旧版还有一个 qlock + WaitQueue waitingProducers/waitingConsumers 的说法。本机 javap -p 确实能看到 ReentrantLock qlock 与两个 WaitQueue 字段，但源码里写得很清楚，它们只为跨版本反序列化而保留：
// JDK 25 源码摘录，已删减
/*
* To cope with serialization across multiple implementation
* overhauls, we declare some unused classes and fields that exist
* solely to enable serializability across versions. These fields
* are never used, so are initialized only if this object is ever
* serialized. We use readResolve to replace a deserialized queue
* with a fresh one.
*/
static class WaitQueue implements java.io.Serializable { }
static class LifoWaitQueue extends WaitQueue { }
static class FifoWaitQueue extends WaitQueue { }
private ReentrantLock qlock;
private WaitQueue waitingProducers;
private WaitQueue waitingConsumers;
qlock 只在 writeObject 里被创建，readResolve 用 waitingProducers instanceof FifoWaitQueue 恢复公平标志，然后返回一个全新的 SynchronousQueue。传输路径一次都没有读过 qlock，公平模式完全由 fair 布尔值选择 xfer 还是 xferLifo。所以“公平模式靠 qlock 保护两个 WaitQueue”是把 JDK 8 的字段名和 JDK 25 的类结构拼在一起的错误叙述。
边界行为与源码一致：它没有容量，size() 恒为 0，remainingCapacity() 恒为 0，isEmpty() 恒为 true，peek() 恒为 null，contains 恒为 false。offer(e) 只有在恰好有消费者等待时才返回 true。本机实测：
== SynchronousQueue 零容量 ==
size=0 remainingCapacity=0 isEmpty=true peek=null
offer(no consumer) = false
offer(with waiting consumer)=true, identitySame=true
交接的是同一个对象引用，没有复制。公平性实验连跑三轮，每轮先让两个消费者挂起，再依次 put("first")、put("second")：
== SynchronousQueue 公平/非公平的交接顺序 ==
fair=true -> [c1:first, c2:second, c1:first, c2:second, c1:first, c2:second]
fair=false -> [c2:first, c1:second, c2:first, c1:second, c2:first, c1:second]
公平模式三轮都是先到的 c1 拿到 first；非公平模式三轮都是后到的 c2 先拿到，符合 LIFO 的预期。这个顺序差异不是实现细节，而是使用语义：线程池用非公平模式获得时间局部性（刚被唤醒的线程更可能在缓存里），消息传递用公平模式获得可预测的先后顺序。
什么时候不要用 SynchronousQueue：把它当普通队列做缓冲。它没有任何缓冲能力，生产速度略高于消费速度就会让生产者 park；本机 1 对 1 的吞吐探针里，它约 41.9 万 ops/s，ArrayBlockingQueue 约 1226 万 ops/s（负载均值 69，只能看数量级）。正确用法是“线程之间的 rendezvous”：Executors.newCachedThreadPool() 用它把任务直接交给空闲线程，交接不出去就新建线程。用它做业务队列，等于把背压直接传导给上游。
PriorityBlockingQueue：堆与扩容自旋锁
PriorityBlockingQueue 的字段很少：
// JDK 25 源码摘录，已删减
private transient Object[] queue;
private transient int size;
private transient Comparator<? super E> comparator;
private final ReentrantLock lock = new ReentrantLock();
private final Condition notEmpty = lock.newCondition();
private transient volatile int allocationSpinLock;
private PriorityQueue<E> q; // 仅序列化兼容用
queue 是一个二叉堆数组：queue[n] 的两个孩子是 queue[2*n+1] 与 queue[2*(n+1)]，父节点不大于孩子，所以 queue[0] 是堆顶。offer 在锁内 siftUp 到合适位置，take 在锁内取走 queue[0] 再把最后一个元素 siftDown。put 直接调用 offer，因为它是无界的，永远不需要等空位；remainingCapacity() 恒返回 Integer.MAX_VALUE。
真正特别的是扩容。tryGrow 不在持有主锁的情况下分配数组：
// JDK 25 源码摘录，已删减
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 &&
ALLOCATIONSPINLOCK.compareAndSet(this, 0, 1)) {
try {
int growth = (oldCap < 64)
? (oldCap + 2) // grow faster if small
: (oldCap >> 1);
int newCap = ArraysSupport.newLength(oldCap, 1, growth);
if (queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
if (newArray == null) // back off if another thread is allocating
Thread.yield();
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
这段代码回答了三个“为什么”：
- 为什么先释放主锁：
new Object[newCap]与System.arraycopy可能是毫秒级操作，分配还可能触发 GC。握着主锁做这件事，所有offer/take都会被卡住；先放锁，让其他线程继续在旧数组上操作。 - 为什么需要
allocationSpinLock：主锁已释放，必须保证同一时刻只有一个线程在扩容。它是一个volatile int，用 VarHandle 做 CAS；抢不到的线程Thread.yield()后回到offer的while重试，而不是自旋等待。 - 为什么最后要检查
queue == array：从放锁到重新拿锁之间，别的线程可能已经完成了扩容。只有当前数组仍是自己看到的那个旧数组时，才把新数组装回去；否则丢弃自己分配的结果，避免覆盖别人的扩容。
扩容策略是“小数组每次 +2，大数组每次 +50%”，oldCap < 64 时增长更快，避免小队列频繁扩容。初始容量默认 11（DEFAULT_INITIAL_CAPACITY）。
堆序不等于稳定排序。本机实验先插入优先级 [7,3,9,1,5]，再依次 poll：
== PriorityBlockingQueue 堆序与稳定性 ==
inserted priorities [7,3,9,1,5], polled=[1, 3, 5, 7, 9]
remainingCapacity=2147483647
8 个同优先级元素的出队序号=[0, 7, 6, 5, 4, 3, 2, 1]（不保证 FIFO）
前一行验证了最小堆；后一行说明同优先级元素的相对顺序由 siftUp/siftDown 的具体路径决定，可能完全反转。需要稳定顺序时，必须在比较器里加第二关键字（例如入队序号），不能依赖插入顺序。
还有一个容易踩的边界：iterator() 不保证按优先级遍历，它只是按堆数组的下标顺序走。要按顺序消费，只能用 poll/take，不能 for (E e : queue) 然后假设拿到了有序序列。
什么时候不要用：它无界，生产速度长期高于消费速度时没有背压，最终 OOM；比较器在锁内执行，比较器里做 I/O 或调用外部服务会把整个队列拖死；它没有公平模式，同优先级也不稳定。需要按时间出队用它的小弟 DelayQueue，需要固定容量背压用 7.1 的 ArrayBlockingQueue。
DelayQueue：leader 与 available
DelayQueue 的字段是三个角色：
// JDK 25 源码摘录，已删减
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader;
private final Condition available = lock.newCondition();
q 按 Delayed.getDelay() 排序，堆顶是最早到期的元素。leader 是 Leader-Follower 模式的“当前负责等下一个到期时刻的线程”，available 是唯一的等待条件。take 的循环是理解这个类的关键：
// JDK 25 源码摘录，已删减
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0L)
return q.poll();
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
协议可以拆成四条：
- 堆空：所有消费者在
available上无限期等待，直到offer唤醒。 - 堆顶已到期：直接
poll返回，不等待。 - 堆顶未到期且已有 leader：自己
await()无限期等待。它不设定时器，因为 leader 会在到期时负责唤醒或重新竞争。 - 堆顶未到期且没有 leader：自己成为 leader，用
awaitNanos(delay)只等到堆顶的到期时刻；醒来后在finally里清掉自己的 leader 身份。
这样做的收益是：无论多少消费者，同一时刻只有一个线程在做定时等待。如果每个消费者都 awaitNanos(delay)，N 个线程会有 N 个定时器，到期时一起醒来抢锁，形成惊群；Leader-Follower 把定时等待压缩到一条线程。
offer 也要维护这个协议：在锁内 q.offer(e) 之后，如果 q.peek() == e（新元素成了更早的堆顶），就把 leader 置空并 available.signal()。原因是当前 leader 等的是旧堆顶的较晚时刻，已经不正确了；旧 leader 醒来后发现 leader != thisThread，不会清空别人的身份，而是回到循环重新判断。
本机实验先放三个元素：B 延迟 300 ms、A 延迟 100 ms、C 延迟 200 ms，然后连续 take 三次：
== DelayQueue 到期顺序与 leader ==
take 顺序与相对时刻: A-100ms@103ms C-200ms@205ms B-300ms@303ms
等待中 leader=Thread-13, c1=Thread-13, c2=Thread-14, c1IsLeader=true
两个消费者最终拿到: [c1:only, c2:later]
前一行验证了“按到期时间而不是插入顺序”出队，延迟误差 3–5 ms 来自线程调度与本机负载。后两行是两个消费者同时 take 的场景：等待 150 ms 时读取私有字段 leader，只有 c1 是 leader，c2 在无限期等待；第一个元素到期后 c1 拿走，第二个元素到期后 c2 拿走。读取私有字段需要 --add-opens java.base/java.util.concurrent=ALL-UNNAMED，这里只为观察，不是生产用法。
什么时候不要用：需要周期性任务、固定速率、取消语义时，用 ScheduledThreadPoolExecutor（11.3 会拆它的 DelayedWorkQueue，同样是堆 + leader）；需要持久化、重启后恢复的延迟任务，DelayQueue 只在内存里；需要毫秒级以下精度时，awaitNanos 的唤醒精度受操作系统的定时器与调度影响，本机高负载下 3–5 ms 的偏差是正常量级。还有一个隐蔽的坑：元素入队后如果 getDelay 的返回值发生变化，堆序就失效了，take 可能一直等到错误的时刻。Delayed 实现应当让到期时间在入队后不可变。
三种队列的边界对照
| 边界 | SynchronousQueue | PriorityBlockingQueue | DelayQueue |
|---|---|---|---|
put 会阻塞吗 | 会，直到有配对者 | 不会，无界 | 不会，无界 |
offer 何时失败 | 无等待消费者时立即 false | 永不失败（除非 OOM） | 永不失败（除非 OOM） |
size() 有意义吗 | 恒 0 | 是元素数 | 是元素数 |
| 顺序由什么决定 | 公平标志 + 到达顺序 | 比较器/自然序 | getDelay |
| 典型误用 | 当缓冲队列 | 当有界队列 | 当高吞吐定时器 |
在 libraryd-conc 里落地
预约到书后要给会员保留 48 小时，超时自动释放库存。这是 DelayQueue 的典型场景：每个预约有自己的到期时刻，消费者只关心“最早到期的那一个”，不需要周期任务。
package com.plume.conc.core;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/** 预约到期队列：到期后由工作线程释放库存。 */
public final class ReservationExpiryQueue {
public record Reservation(String loanId, String isbn, String memberId, Instant dueAt) {}
public record ExpiringReservation(Reservation reservation) implements Delayed {
public long getDelay(TimeUnit unit) {
long nanos = Duration.between(Instant.now(), reservation.dueAt()).toNanos();
return unit.convert(nanos, TimeUnit.NANOSECONDS);
}
public int compareTo(Delayed other) {
return reservation.dueAt().compareTo(
((ExpiringReservation) other).reservation.dueAt());
}
}
private final DelayQueue<ExpiringReservation> queue = new DelayQueue<>();
public void schedule(Reservation reservation) {
queue.put(new ExpiringReservation(reservation));
}
/** 阻塞到最早一个预约到期；调用方负责释放库存。 */
public Reservation awaitExpired() throws InterruptedException {
return queue.take().reservation();
}
public int pending() {
return queue.size();
}
}
ExpiringReservation 的 compareTo 必须与 getDelay 的排序一致，都按 dueAt；如果 compareTo 用了别的字段，堆顶就不一定是最早到期者，take 会等到错误的时间。schedule 里的 put 不会阻塞，但队列无界，长期堆积说明释放库存的消费者停了，需要靠 pending() 做监控。
第二个落点是直接交接。馆内自助机的借书请求可以交给固定的几个虚拟线程处理，交接不出去就立即拒绝，而不是排队：
package com.plume.conc.core;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
/** 零容量交接：没有空闲处理者时立刻返回 false。 */
public final class HandoffDispatcher {
private final SynchronousQueue<Runnable> handoff = new SynchronousQueue<>(true);
public boolean dispatch(Runnable task, long timeoutMillis) throws InterruptedException {
return handoff.offer(task, timeoutMillis, TimeUnit.MILLISECONDS);
}
}
这里用 offer(timeout) 而不是 put：自助机不能因为后端处理者全忙就无限挂起，超时后返回 false，前端提示“请稍后再试”。如果改成 put，高峰时所有请求线程都会 park 在 SynchronousQueue 上，等于把线程池的排队问题转移到了调用线程。
常见错误与边界
- 用
size()监控SynchronousQueue的积压：它恒为 0，要看交接延迟或offer失败率；调用put而没有配对者会永久阻塞，优先用带超时的offer/poll。 - 把
PriorityBlockingQueue当有界队列：remainingCapacity()恒为Integer.MAX_VALUE，没有背压。 - 用
for (E e : priorityQueue)期待有序输出：迭代器按堆数组下标走，不按优先级；在DelayQueue元素入队后修改到期时间会让堆序失效。 - 把
DelayQueue当高频定时器：所有操作共用一把锁，每次循环还要调用getDelay；高吞吐场景应换时间轮或专用调度器。
小结
SynchronousQueue 在 JDK 25 已经统一到 LinkedTransferQueue：Transferer 继承它，非公平模式加一个 xferLifo，公平模式复用父类的 FIFO xfer；qlock/WaitQueue 只是反序列化兼容字段，不参与交接。PriorityBlockingQueue 用二叉堆提供无界的优先级出队，扩容时释放主锁、用 allocationSpinLock 做 CAS 分配，避免在分配和复制期间卡住所有操作。DelayQueue 用 PriorityQueue 加 Leader-Follower，把定时等待压缩到一个 leader 线程，其他消费者无限期等待。三者都不是“更好的队列”，而是三种不同的等待条件：等配对、等堆顶、等到期。
7.3 把这三个类和 7.1 的两个类放在一起，比较锁粒度、节点开销、伪共享与容量语义，并解释为什么本机测出的吞吐数字只能看数量级。
阅读导航：上一节 7.1 ArrayBlockingQueue 与 LinkedBlockingQueue ｜ 下一节 7.3 队列的锁粒度、吞吐与内存
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。