承接的问题
卷一第 11 章讲的是 ScheduledExecutorService 的用法：scheduleAtFixedRate 与 scheduleWithFixedDelay 的差别、周期任务漂移怎么来的。本节看容器：ScheduledThreadPoolExecutor 内部那个没有公开的 DelayedWorkQueue。
要回答三个问题：到期任务按时间排序，为什么不用 PriorityQueue 而自己写一个数组堆？取消一个尚未到期的任务，凭什么能做到 O(log n)？多个 worker 同时等在同一个到期任务上，是谁去睡、睡多久？
数组堆与 heapIndex 反查
DelayedWorkQueue 是一个 AbstractQueue<Runnable>，内部是数组堆加一把 ReentrantLock：
// JDK 25 源码摘录，已删减
private static final int INITIAL_CAPACITY = 16;
private RunnableScheduledFuture<?>[] queue =
new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
private final ReentrantLock lock = new ReentrantLock();
private int size;
private Thread leader;
private final Condition available = lock.newCondition();
private static void setIndex(RunnableScheduledFuture<?> f, int idx) {
if (f instanceof ScheduledFutureTask)
((ScheduledFutureTask)f).heapIndex = idx;
}
关键在 setIndex：每个 ScheduledFutureTask 自己记着"我在堆数组的哪个下标"。siftUp / siftDown 每搬动一个元素就调一次 setIndex，所以堆序变化之后反查表仍然准确。取消时不必扫描数组：
// JDK 25 源码摘录，已删减
private int indexOf(Object x) {
if (x != null) {
if (x instanceof ScheduledFutureTask) {
int i = ((ScheduledFutureTask) x).heapIndex;
// Sanity check; x could conceivably be a
// ScheduledFutureTask from some other pool.
if (i >= 0 && i < size && queue[i] == x)
return i;
} else {
for (int i = 0; i < size; i++)
if (x.equals(queue[i]))
return i;
}
}
return -1;
}
两处防御值得注意。第一，queue[i] == x 用引用相等而不是 equals，因为下标可能来自另一个池的同名任务，或者已经过期失效。第二，只有 ScheduledFutureTask 有 heapIndex，别的 RunnableScheduledFuture 实现只能退回线性扫描——源码注释的原文是"we expect that most tasks will not be decorated, and that the faster cases will be much more common"。
remove 拿到下标后做的是"末尾元素补位、先 siftDown 再按需 siftUp"，并把被删任务的 heapIndex 置 -1。取消的复杂度因此从 O(n) 降到 O(log n)，同时避免了一个更隐蔽的问题：如果只标记不摘除，已取消的任务会一直占着堆里的位置，直到它升到堆顶才被清理。扩容则是每次加 50%：
// JDK 25 源码摘录，已删减
private void grow() {
int oldCapacity = queue.length;
int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
if (newCapacity < 0) // overflow
newCapacity = Integer.MAX_VALUE;
queue = Arrays.copyOf(queue, newCapacity);
}
下面这段程序直接拿一个真实的 ScheduledThreadPoolExecutor 观察队列：先看 leader 字段和任务实际跑在哪个线程上，再看扩容节奏、DelayQueue 的对比容量，最后看 heapIndex 在取消前后的变化。
import java.lang.reflect.Field;
import java.util.PriorityQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/** 观察 ScheduledThreadPoolExecutor 内部 DelayedWorkQueue 的 leader、扩容与 heapIndex。 */
public final class QueueProbe {
record D(long trigger) implements Delayed {
public long getDelay(TimeUnit u) {
return u.convert(trigger - System.nanoTime(), TimeUnit.NANOSECONDS);
}
public int compareTo(Delayed o) {
return Long.compare(getDelay(TimeUnit.NANOSECONDS), o.getDelay(TimeUnit.NANOSECONDS));
}
}
public static void main(String[] args) throws Exception {
AtomicInteger seq = new AtomicInteger();
ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(4,
r -> Thread.ofPlatform().name("sched-" + seq.incrementAndGet()).unstarted(r));
stpe.prestartAllCoreThreads();
stpe.setRemoveOnCancelPolicy(false);
Object q = stpe.getQueue();
Field leaderF = q.getClass().getDeclaredField("leader");
leaderF.setAccessible(true);
Field arrayF = q.getClass().getDeclaredField("queue");
arrayF.setAccessible(true);
Thread.sleep(200);
System.out.println("4 个 worker 空闲时 leader=" + leaderF.get(q));
stpe.schedule(() -> System.out.println("T1 跑在 " + Thread.currentThread().getName()),
300, TimeUnit.MILLISECONDS);
Thread.sleep(20);
Object leader = leaderF.get(q);
System.out.println("schedule 之后 leader="
+ (leader == null ? "null（尚未交接）" : ((Thread) leader).getName()));
Thread.sleep(1000);
System.out.println("1 秒后 size=" + stpe.getQueue().size() + " poolSize=" + stpe.getPoolSize());
for (int i = 2; i <= 4; i++) {
int id = i;
stpe.schedule(() -> System.out.println("T" + id + " 跑在 "
+ Thread.currentThread().getName()), 0, TimeUnit.MILLISECONDS);
}
Thread.sleep(500);
int last = 0;
for (int i = 0; i < 40; i++) {
stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
int cap = ((Object[]) arrayF.get(q)).length;
if (cap != last) {
System.out.println("DelayedWorkQueue 元素数=" + stpe.getQueue().size() + " 数组长度=" + cap);
last = cap;
}
}
DelayQueue<D> dq = new DelayQueue<>();
for (int i = 0; i < 40; i++) dq.add(new D(System.nanoTime() + TimeUnit.SECONDS.toNanos(60)));
Field pqField = DelayQueue.class.getDeclaredField("q");
pqField.setAccessible(true);
Field pqArray = PriorityQueue.class.getDeclaredField("queue");
pqArray.setAccessible(true);
System.out.println("DelayQueue 元素数=" + dq.size()
+ " 内部数组长度=" + ((Object[]) pqArray.get(pqField.get(dq))).length);
Field heapIndexF = Class.forName(
"java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask")
.getDeclaredField("heapIndex");
heapIndexF.setAccessible(true);
ScheduledFuture<?> f1 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
ScheduledFuture<?> f2 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
ScheduledFuture<?> f3 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
System.out.println("三个任务 heapIndex=" + heapIndexF.get(f1) + ","
+ heapIndexF.get(f2) + "," + heapIndexF.get(f3));
System.out.println("removeOnCancel=false: cancel 中间一个 -> " + f2.cancel(false)
+ " heapIndex=" + heapIndexF.get(f2) + " 队列 size=" + stpe.getQueue().size());
stpe.setRemoveOnCancelPolicy(true);
ScheduledFuture<?> g1 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
ScheduledFuture<?> g2 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
ScheduledFuture<?> g3 = stpe.schedule(() -> { }, 60, TimeUnit.SECONDS);
System.out.println("removeOnCancel=true: 三个任务 heapIndex=" + heapIndexF.get(g1) + ","
+ heapIndexF.get(g2) + "," + heapIndexF.get(g3));
System.out.println("cancel 中间一个 -> " + g2.cancel(false)
+ " heapIndex=" + heapIndexF.get(g2) + " 队列 size=" + stpe.getQueue().size());
stpe.shutdownNow();
}
}
用 javac --release 25 编译，再以 --add-opens java.base/java.util.concurrent=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED 运行，本机输出：
4 个 worker 空闲时 leader=null
schedule 之后 leader=sched-1
T1 跑在 sched-1
1 秒后 size=0 poolSize=4
T2 跑在 sched-2
T3 跑在 sched-4
T4 跑在 sched-3
DelayedWorkQueue 元素数=1 数组长度=16
DelayedWorkQueue 元素数=17 数组长度=24
DelayedWorkQueue 元素数=25 数组长度=36
DelayedWorkQueue 元素数=37 数组长度=54
DelayQueue 元素数=40 内部数组长度=50
三个任务 heapIndex=40,41,42
removeOnCancel=false: cancel 中间一个 -> true heapIndex=41 队列 size=43
removeOnCancel=true: 三个任务 heapIndex=43,44,45
cancel 中间一个 -> true heapIndex=-1 队列 size=45
先看后半段的容量与取消：
- 16 → 24 → 36 → 54，每次加 50%，与
grow()的oldCapacity + (oldCapacity >> 1)一致。DelayQueue的PriorityQueue起始 11，小数组时按old + old + 2翻倍，40 个元素后是 50。 heapIndex是从 40 开始的，因为堆里前面已经排了 40 个 60 秒后到期的任务；三个新任务接在末尾，正是"新元素先放到数组尾部、再siftUp“的表现。- 默认
removeOnCancel=false时cancel返回 true，但heapIndex还是 41、size还是 43。 任务只是被标记为已取消，仍占着堆里的位置，要等到它的到期时间点才被run()发现并丢弃。长延迟任务被大量取消时，队列会持续膨胀。 - 开启
setRemoveOnCancelPolicy(true)后heapIndex变成 -1、size从 46 变 45。 这才是真正走remove→indexOf→siftDown的反查路径，heapIndex被setIndex(f, -1)清掉，正是"移除"的标志。
leader 与 available：只让一个人等
take() 是整个类里最值得逐分支读的方法：
// JDK 25 源码摘录，已删减
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0];
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0L)
return finishPoll(first);
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 && queue[0] != null)
available.signal();
lock.unlock();
}
}
四个分支：
- 队列空 →
available.await()无限期等，直到有人offer。 - 堆顶已到期（
delay <= 0）→finishPoll摘掉堆顶，返回。 - 有 leader 了 → 自己
available.await()无限期等，不设超时。 - 没有 leader → 自己当 leader，
awaitNanos(delay)只睡到堆顶的到期时刻。
这就是 Leader-Follower 的变体：同一时刻只有一个线程带超时睡，其余线程无限期睡，等 leader 醒来处理完再交接。 如果让所有人都 awaitNanos(delay)，每次定时等待都要挂一个定时器，到期时全部唤醒、全部抢锁、全部发现"任务已经被别人拿走了”，白白制造一次惊群。只让一个人带超时，其余人靠 signal 唤醒，唤醒次数就从"等待者数量"降到"任务数量"。first = null 那行也不是随手写的：等待期间不再持有堆顶任务的引用，避免一个大任务对象被等待者拖住无法回收。
交接发生在两处。take 的 finally 里：
// JDK 25 源码摘录，已删减
if (leader == null && queue[0] != null)
available.signal();
离开时如果自己已经不是 leader（或当 leader 期间被 finally 清掉了），而队列里还有下一个任务，就叫醒一个等待者去当新的 leader。offer 里则是另一头：
// JDK 25 源码摘录，已删减
if (queue[0] == e) {
leader = null;
available.signal();
}
新任务插到了堆顶（比原堆顶更早到期），原来的 leader 睡过头了：把 leader 置空作废，signal 叫醒一个线程重新评估。被叫醒的线程可能发现自己抢不到 leadership（别人先到），于是回到 await()——源码注释专门提醒"waiting threads must be prepared to acquire and lose leadership while waiting"。
上面的输出正好是这套机制的现场：四个 worker 空闲时 leader=null；排入一个 300 ms 后到期的任务后，sched-1 成为 leader，也只有它被定时唤醒，T1 跑在 sched-1；另外三个仍在 available.await() 上无限期等，直到补进三个零延迟任务，才各自被 signal 唤醒，T2/T3/T4 分别跑在 sched-2/sched-4/sched-3 上（谁拿到哪个由唤醒顺序决定，重复运行会变，但每个任务只被取走一次这一点不变）。1 秒后 size=0 说明到期任务已被摘除，没有重复投递。
与 DelayQueue 的差异
两者都是"堆 + leader + available"，差异在于 DelayQueue 把堆交给 PriorityQueue，DelayedWorkQueue 自己管数组：
| 维度 | DelayedWorkQueue | DelayQueue |
|---|---|---|
| 底层容器 | 自管 RunnableScheduledFuture[] 数组 | PriorityQueue<E> |
| 初始容量 | INITIAL_CAPACITY = 16 | 11（PriorityQueue 默认） |
| 扩容 | 每次 +50%：16→24→36→54 | 小于 64 时 old+old+2，之后 +50%；实测 40 个元素为 50 |
| 删除定位 | ScheduledFutureTask.heapIndex 反查，O(log n) | 线性扫描 + equals，O(n) |
| 元素类型 | 必须是 RunnableScheduledFuture（offer 强转） | 任意 Delayed |
| 取消即移除 | 需 setRemoveOnCancelPolicy(true) | remove 立即摘除 |
ScheduledThreadPoolExecutor 不用 DelayQueue 的核心原因就是删除定位：定时任务的生命周期往往由业务侧控制（请求超时、会话过期、订单关闭），取消是高频操作；用 DelayQueue 每次取消都要 O(n) 扫描，还要依赖元素的 equals 实现正确。代价是耦合——offer 会把参数强转成 RunnableScheduledFuture，传别的 Delayed 直接 ClassCastException；heapIndex 只对 ScheduledFutureTask 生效，别的实现照样退回线性扫描。
在 libraryd-conc 里落地
借阅到期提醒天然适合定时任务：借出时排一个到期任务，归还时取消。落地时最容易踩的坑就是"取消了但没摘除"：
package com.plume.conc.core;
import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/** 借阅到期提醒。取消时必须真的把任务从堆里摘掉。 */
public final class DueReminder implements AutoCloseable {
private final ScheduledThreadPoolExecutor scheduler;
private final Map<String, ScheduledFuture<?>> pending = new ConcurrentHashMap<>();
public DueReminder(int threads) {
this.scheduler = new ScheduledThreadPoolExecutor(threads,
r -> Thread.ofPlatform().name("due-reminder-", 0).unstarted(r));
this.scheduler.setRemoveOnCancelPolicy(true); // 关键：取消即摘除
this.scheduler.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
}
public void schedule(String loanId, Instant dueAt, Runnable onDue) {
long delay = Math.max(0, Duration.between(Instant.now(), dueAt).toMillis());
ScheduledFuture<?> old = pending.put(loanId,
scheduler.schedule(onDue, delay, TimeUnit.MILLISECONDS));
if (old != null) old.cancel(false); // 续借：先排新的再取消旧的
}
public boolean cancel(String loanId) {
ScheduledFuture<?> f = pending.remove(loanId);
return f != null && f.cancel(false);
}
public int pendingTasks() {
return scheduler.getQueue().size();
}
@Override
public void close() {
scheduler.shutdownNow();
}
}
三个要点。第一，setRemoveOnCancelPolicy(true) 必须显式打开——默认 false，取消只打标记，堆里仍占位，pendingTasks() 会长期高于真实待办数（上面的实测里，cancel 之后 size 一点没变）。第二，续借要"先排新的、再取消旧的"，反过来会有一个短暂窗口没有待办任务。第三，setExecuteExistingDelayedTasksAfterShutdownPolicy(false) 让 shutdown 之后的到期任务不再执行，配合 close() 里的 shutdownNow()，停机时不会突然弹出一批提醒。
这个设计的代价
第一，一把锁管所有操作。 take、offer、poll、size、contains 全部抢同一把 ReentrantLock，size() 也不例外（isEmpty() 直接调 size()）。定时任务量大、周期短时，这把锁就是吞吐上限。卷一第 11 章讲过"大量短周期任务用时间轮"，从实现角度看就是躲开这把锁和每次堆调整。
第二，数组只扩不缩。 grow 没有对应的 shrink，一次高峰把数组撑大之后，即使队列长期为空也不会归还。长生命周期的调度器要定期 purge() 清理已取消任务，或者按业务拆成多个调度器。
第三，leader 是单点状态。 它保证只有一个线程带超时等待，代价是堆顶变化必须作废它（offer 里 leader = null），被 signal 唤醒的线程还要重新检查自己是否还能当 leader；高竞争下表现为几次多余的唤醒和抢锁。
第四，不要拿它当通用延迟队列。 它只接受 RunnableScheduledFuture，offer 的强转不做检查；iterator 只是创建时的数组快照，遍历期间的增删不可见。要通用延迟队列用 DelayQueue，要定时执行用 ScheduledThreadPoolExecutor，两者不要互相替代。
小结
DelayedWorkQueue 在"堆 + leader"的骨架上做了两处改造：把 PriorityQueue 换成自管数组，让每个 ScheduledFutureTask 记住自己的 heapIndex，使取消从 O(n) 扫描变成 O(log n) 反查；用 leader 保证同一时刻只有一个线程带超时等待，其余线程无限期 await，靠 signal 逐个交接。本机实测确认了 16→24→36→54 的扩容节奏、默认不摘除取消任务的行为，以及四个 worker 等一个到期任务时只有 leader 被定时唤醒。
下一章进入 ForkJoinPool：ctl 换成 long，WorkQueue 用 base/top 两端推进加 phase seqlock，旧教材里的 scan / steal 在 JDK 25 里已经没有同名方法。
阅读导航：上一节 11.2 addWorker、runWorker 与退出 ｜ 下一节 12.1 工作窃取队列与任务扫描
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。