调度与限流 #03：最小堆定时器

最小堆定时器以 PriorityQueue 为核心，堆顶始终是最近到期任务，插入 O(log N)、取消 O(log N)；JDK ScheduledThreadPoolExecutor 的 DelayedWorkQueue 以及几乎所有操作系统内核早期定时器均基于此结构。

相关文章：时间轮（Hashed Timing Wheel） · 层级时间轮

目录

问题背景

什么场景需要最小堆定时器？

最小堆定时器是最自然、最通用的定时任务调度结构：

• 任务数量少（< 10 万级别），delay 跨度极大（毫秒到天级别混用）
• 精度要求高：每个任务有独立的 deadline，无需为精度牺牲内存或 CPU
• 实现简单：JDK 标准库 PriorityQueue 即可驱动

与时间轮的核心矛盾

时间轮用空间换时间，把任务分散到固定数量的槽；最小堆用时间换精度，每次维护全局有序。

两者适用场景天然互补：

任务数 × log N  vs  O(1)
  少任务大跨度    vs  海量任务固定精度

核心数据结构

// 定时器任务
struct TimerTask {
    long      deadline       // 绝对到期时间戳（单调时钟纳秒，不受时钟回拨影响）
    Runnable  callback       // 任务逻辑
    boolean   cancelled      // 惰性取消标志（volatile，保证可见性）
    int       heapIndex      // 在堆数组中的下标（支持 O(log N) cancel）
                             // 不持有该字段则 cancel 只能 O(1) 标记，无法立即从堆中删除
    long      period         // 周期任务间隔（ms），0 表示一次性任务
}

// 最小堆定时器
struct MinHeapTimer {
    TimerTask[]  heap         // 动态数组，1-indexed（heap[1] 为堆顶）
    int          size         // 当前堆元素数量
    long         tickResolution  // 后台线程唤醒精度（ms），通常 1ms~100ms
    Thread       timerThread  // 单一后台线程，负责 tick 推进
    Executor     taskExecutor // 任务回调的执行线程池（与 timerThread 分离）
}

// 堆序不变式
// 对所有 i ∈ [1, size/2]：
//   heap[i].deadline <= heap[2*i].deadline        （父 ≤ 左子）
//   heap[i].deadline <= heap[2*i+1].deadline      （父 ≤ 右子）
// 即：堆顶 heap[1] 是 deadline 最小（最快到期）的任务

关键设计决策：

• heapIndex 字段：额外占 4 字节，换来 cancel 时 O(log N) 直接定位，无需 O(N) 遍历
• 单调时钟：deadline 使用 System.nanoTime()（单调递增），不受 NTP 调整影响，防止时钟回拨导致任务漏触发
• timerThread 与 taskExecutor 分离：timerThread 只负责时钟推进，任务实际执行交给线程池，避免耗时任务阻塞时钟

schedule() 伪代码

// 向定时器插入一个延迟 delayMs 后触发的任务
// 时间复杂度：O(log N)，N = 当前堆中任务数
function schedule(callback, delayMs):
    task = new TimerTask()
    task.deadline  = nanoTime() + delayMs * 1_000_000   // 转为纳秒
    task.callback  = callback
    task.cancelled = false
    task.period    = 0    // 一次性任务

    heap.add(task)                          // 追加到数组末尾
    task.heapIndex = size                   // 记录当前下标
    siftUp(size)                            // 上浮维护堆序
    size++

    if task.heapIndex == 1:                 // 新任务成了堆顶（更早到期）
        timerThread.wakeUp()               // 唤醒后台线程重新计算等待时间

    return task    // 返回句柄，供 cancel() 使用

// 上浮操作（sift-up）
// 将 heap[i] 向上移动，直到满足堆序
function siftUp(i):
    while i > 1:
        parent = i / 2
        if heap[parent].deadline <= heap[i].deadline:
            break       // 父节点已 ≤ 当前节点，堆序满足，停止
        swap(heap[i], heap[parent])
        heap[i].heapIndex      = i
        heap[parent].heapIndex = parent
        i = parent

插入步骤拆解：

1. 创建 TimerTask，设置 deadline = now + delay，O(1)
2. 数组末尾追加，记录 heapIndex，O(1)
3. siftUp：最多上浮 log₂(N) 层，每层一次比较 + 最多一次 swap，O(log N)
4. 若新任务成为堆顶（deadline 最小），唤醒后台线程重设等待时间，O(1)

cancel() 伪代码

方案一：惰性删除（lazy cancel）

// O(1) 标记取消，不立即从堆中删除
function cancelLazy(task):
    task.cancelled = true   // CAS 原子设置（防止重复 cancel 与并发 fire 竞争）

// tick() 中检查
if task.cancelled:
    heap.poll()    // 弹出但不执行
    continue

优点：实现最简单，cancel 路径完全无锁
缺点：堆中可能积压大量"僵尸任务"，浪费内存；堆操作仍需比较这些节点

方案二：立即删除（eager cancel，需要 heapIndex）

// O(log N) 立即从堆中移除
// 前提：TimerTask 持有 heapIndex 字段
function cancelEager(task):
    if task.heapIndex < 1 or task.heapIndex > size:
        return   // 任务已执行或已删除

    i = task.heapIndex

    // 用堆末尾元素覆盖被删元素，然后缩小堆
    heap[i] = heap[size]
    heap[i].heapIndex = i
    size--

    // 新填入的元素可能需要上浮或下沉（只会走其中一条路）
    siftUp(i)
    siftDown(i)

    task.heapIndex = -1    // 标记为已移除

// 下沉操作（sift-down）
function siftDown(i):
    while true:
        smallest = i
        left     = 2 * i
        right    = 2 * i + 1

        if left <= size and heap[left].deadline < heap[smallest].deadline:
            smallest = left
        if right <= size and heap[right].deadline < heap[smallest].deadline:
            smallest = right

        if smallest == i:
            break    // 已满足堆序
        swap(heap[i], heap[smallest])
        heap[i].heapIndex       = i
        heap[smallest].heapIndex = smallest
        i = smallest

两种方案对比：

tick() 伪代码

单线程执行模型

// timerThread 的主循环（单线程，负责时钟推进）
function tickLoop():
    while running:
        now = nanoTime()

        // 触发所有已到期任务（deadline <= now）
        while size > 0 and heap[1].deadline <= now:
            task = heap[1]
            removeTop()          // 弹出堆顶，O(log N)
            if not task.cancelled:
                taskExecutor.submit(task.callback)   // 异步提交，不阻塞 timerThread

        // 计算下次唤醒时间
        if size > 0:
            sleepNs = heap[1].deadline - nanoTime()
            if sleepNs > 0:
                timerThread.sleep(sleepNs)    // 精确等到堆顶到期
        else:
            timerThread.wait()               // 无任务，等待 schedule() 唤醒

// 弹出堆顶（sift-down）
function removeTop():
    heap[1] = heap[size]
    heap[1].heapIndex = 1
    size--
    siftDown(1)

多线程执行模型注意事项

// 多线程 schedule/cancel 需加锁（或使用无锁结构）
lock.acquire()
try:
    schedule(callback, delay)    // 修改 heap 数组必须串行
finally:
    lock.release()

// tick() 中弹出堆顶也需持锁
lock.acquire()
try:
    task = heap[1]
    removeTop()
finally:
    lock.release()
task.callback.run()   // 实际执行在锁外，避免持锁时阻塞

关键原则：

• timerThread 只推进时钟，不执行任务：任何耗时操作都必须提交给 taskExecutor
• sleep 精度：Thread.sleep(ns) 精度受 OS 调度影响（Linux 约 ±0.5ms），实际误差通常远小于 tickDuration
• 新任务提前唤醒：若新插入任务 deadline < 当前等待的 deadline，必须打断 sleep 重设等待时间

执行追踪

配置：空堆初始状态，当前时间 t=0

操作序列：

t=0ms: schedule(A, delay=5s)   → A.deadline = 5000ms
t=0ms: schedule(B, delay=2s)   → B.deadline = 2000ms
t=0ms: schedule(C, delay=8s)   → C.deadline = 8000ms

步骤一：插入 A(deadline=5000)

heap 初始为空，直接追加：
  heap[1] = A(5000)
  size = 1

siftUp(1)：i=1，无父节点，退出

堆状态：
         A(5000)
         [1]

步骤二：插入 B(deadline=2000)

追加到末尾：
  heap[2] = B(2000)
  size = 2

siftUp(2)：
  i=2, parent=1
  heap[1].deadline=5000 > heap[2].deadline=2000
  → swap(heap[2], heap[1])：B 上浮到堆顶
  → heap[1]=B(2000), heap[1].heapIndex=1
  → heap[2]=A(5000), heap[2].heapIndex=2
  i=1，退出循环

堆状态：
         B(2000)
        /
       A(5000)
       [1]   [2]

B 成为新堆顶（比 A 更早到期）→ 唤醒 timerThread 重设 sleep 为 2000ms

步骤三：插入 C(deadline=8000)

追加到末尾：
  heap[3] = C(8000)
  size = 3

siftUp(3)：
  i=3, parent=1
  heap[1].deadline=2000 ≤ heap[3].deadline=8000
  → 满足堆序，退出

堆状态（数组下标）：
         B(2000)      ← heap[1]，堆顶
        /       \
    A(5000)   C(8000)  ← heap[2], heap[3]

步骤四：t=2000ms，tickLoop 触发

now = 2000ms
heap[1].deadline = 2000 ≤ 2000 → 触发

removeTop()：
  heap[1] = heap[3] = C(8000)，size=2
  siftDown(1)：
    i=1, left=2(A,5000), right=3(越界 size=2)
    smallest = min(C=8000, A=5000) → A 更小
    swap(heap[1], heap[2])
    heap[1]=A(5000), heap[2]=C(8000)
    i=2, left=4(越界)→ 退出

taskExecutor.submit(B.callback)   ← B 被异步执行

堆状态：
         A(5000)      ← heap[1]，新堆顶
        /
    C(8000)           ← heap[2]

timerThread 重设 sleep = 5000 - 2000 = 3000ms

步骤五：t=5000ms，触发 A；t=8000ms，触发 C

t=5000ms：
  heap[1]=A(5000)，deadline ≤ now → removeTop()
  taskExecutor.submit(A.callback)
  堆：heap[1]=C(8000)，size=1

t=8000ms：
  heap[1]=C(8000)，deadline ≤ now → removeTop()
  taskExecutor.submit(C.callback)
  堆：size=0，timerThread 进入 wait() 状态

JDK ScheduledThreadPoolExecutor

DelayedWorkQueue：最小堆的工业级实现

ScheduledThreadPoolExecutor 内部使用 DelayedWorkQueue，本质是支持 cancel 的最小堆：

// ScheduledFutureTask 关键字段（简化）
class ScheduledFutureTask<V> implements RunnableScheduledFuture<V> {
    long    time;          // 绝对到期时间（纳秒，System.nanoTime()）
    long    period;        // 0=一次性, >0=fixedRate, <0=fixedDelay
    int     heapIndex;     // 在 DelayedWorkQueue 数组中的下标（支持 O(log N) cancel）
    long    sequenceNumber // 相同 time 时的排序稳定性保证（FIFO）
}

// DelayedWorkQueue 关键操作
class DelayedWorkQueue {
    RunnableScheduledFuture<?>[] queue;   // 最小堆数组（初始容量 16，动态扩容）
    int size;
    ReentrantLock lock;
    Condition available;   // 用于 leader-follower 等待

    // cancel 时从堆中移除（利用 heapIndex 直接定位，O(log N)）
    boolean remove(Object x) {
        int i = ((ScheduledFutureTask)x).heapIndex;
        if (i < 0) return false;
        setIndex(queue[i], -1);        // 标记已移除
        int s = --size;
        RunnableScheduledFuture<?> replacement = queue[s];
        queue[s] = null;
        if (s != i) {
            siftDown(i, replacement);  // 先尝试下沉
            if (queue[i] == replacement)
                siftUp(i, replacement); // 下沉无效则上浮
        }
        return true;
    }
}

leader-follower 等待模式

问题：多个工作线程同时调用 take()（从堆顶取任务），如果都 sleep 等到堆顶到期，会引发惊群效应（大量线程同时唤醒，只有一个能拿到任务）。

解决方案：leader-follower 模式

// DelayedWorkQueue.take() 简化逻辑
function take():
    lock.acquire()
    loop:
        first = heap[1]    // 堆顶任务

        if first == null:
            available.await()    // 无任务，无限等待，成为 follower

        delay = first.deadline - nanoTime()

        if delay <= 0:
            return poll()        // 任务已到期，直接取走

        if leader != null:
            // 已有 leader 在等待堆顶，当前线程作为 follower 无限等待
            available.await()
        else:
            // 当前线程成为 leader，精确 sleep 到堆顶到期
            thisThread = Thread.currentThread()
            leader = thisThread
            try:
                available.awaitNanos(delay)    // 精确等待
            finally:
                if leader == thisThread:
                    leader = null

    // leader 醒来后，通知 follower（follower 再竞争成为新 leader）
    if leader == null and heap[1] != null:
        available.signal()
    lock.release()

效果：任意时刻只有 1 个线程（leader）在计时等待，其余线程（follower）无限等待信号，消除惊群效应。

周期任务的重调度

// ScheduledFutureTask.run() 简化
void run() {
    boolean periodic = isPeriodic();
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    else if (!periodic)
        super.run();             // 一次性任务，直接执行
    else if (super.runAndReset()) {
        setNextRunTime();        // 重设下次触发时间
        reExecutePeriodic(outerTask);   // 重新插入堆
    }
}

void setNextRunTime() {
    long p = period;
    if (p > 0)    time += p;             // fixedRate：基于上次触发时间累加
    else          time = nanoTime() - p; // fixedDelay：基于上次完成时间累加（p<0）
}

异常与边界场景

场景 1：系统时钟回拨

现象：NTP 同步导致 System.currentTimeMillis() 向前跳跃（如从 t=10s 突然回到 t=9s），基于此计算的 deadline 可能导致任务被提前触发或错过触发。

算法处理：

// 错误做法（受时钟回拨影响）
task.deadline = System.currentTimeMillis() + delayMs   // ❌

// 正确做法（单调时钟）
task.deadline = System.nanoTime() + delayMs * 1_000_000  // ✅
// System.nanoTime() 保证单调递增，不受 NTP 调整影响
// JDK ScheduledThreadPoolExecutor 使用 System.nanoTime()

即使 nanoTime 本身在 JVM 重启后可能不连续，但在同一 JVM 生命周期内单调性有保证。

场景 2：任务执行抛出异常

现象：callback.run() 抛出 RuntimeException，是否影响后续任务触发？

算法处理：

// tickLoop 中提交给 taskExecutor（线程池）
taskExecutor.submit(task.callback)  // 异步，不阻塞 timerThread

// 线程池内部（ThreadPoolExecutor 的 runWorker）
try:
    task.run()
catch Throwable t:
    // 异常被线程池捕获，不传播到 timerThread
    // afterExecute(task, t) 回调可用于监控

结论：任务异常不影响后续任务，因为 timerThread 与 taskExecutor 完全分离。

但注意 ScheduledThreadPoolExecutor 中的周期任务：若周期任务抛出未检异常，runAndReset() 返回 false，任务不会被重新调度——相当于静默停止。需在任务内部 try-catch 所有异常。

场景 3：并发 cancel 与 fire 竞争

现象：timerThread 正在执行 removeTop() 准备提交任务，同时另一线程调用 cancelEager(task)。

竞态分析：

线程 A（timerThread）：
  task = heap[1]           // 取出堆顶引用
  removeTop()              // 从堆中删除
  // [竞态窗口]
  if not task.cancelled:   // 检查取消标志
      taskExecutor.submit(task.callback)

线程 B（业务线程）：
  task.cancelled = true    // CAS 设置（volatile write）
  cancelEager(task)        // 尝试从堆中删除（此时任务可能已被 removeTop 移出）

关键：task.cancelled 必须声明为 volatile，保证跨线程可见性。cancelEager 中检查 heapIndex == -1 可安全忽略已移出的任务。

竞态结果有两种：

1. cancel 先于 fire：task.cancelled=true → timerThread 跳过提交，任务不执行（正确）
2. fire 先于 cancel：removeTop() 已完成，cancelEager 发现 heapIndex=-1 静默返回；若 task.callback 已提交线程池，cancel 无法阻止执行（best-effort 语义，与 Future.cancel(false) 一致）

场景 4：堆满时的处理

现象：持续调用 schedule() 而任务执行速度不及调度速度，heap 数组超出内存或达到设定上限。

算法处理：

// 方案一：动态扩容（JDK PriorityQueue 默认）
if size == heap.length:
    // 类似 ArrayList：容量 < 64 时翻倍，≥ 64 时增加 50%
    heap = Arrays.copyOf(heap, newCapacity)
// 扩容代价 O(N)，分摊到每次插入仍是 O(log N)
// 风险：OOM — 无边界增长

// 方案二：固定上限 + 拒绝策略（Netty HashedWheelTimer 的 pendingTimeouts 对标）
if size >= maxPendingTasks:
    throw new RejectedExecutionException("timer queue overflow, size=" + size)

// 方案三：惰性清理（减少堆中"僵尸"任务占用的空间）
// 在 tick() 中顺带移除 cancelled 任务，降低峰值堆大小
while size > 0 and heap[1].cancelled:
    removeTop()   // O(log N)

与时间轮对比

选型结论：

任务数 < 10 万 且 delay 跨度极大（ms ~ 天混用）
    → 最小堆（ScheduledThreadPoolExecutor）

任务数 > 10 万 且 delay 相对固定（如统一超时 30s）
    → 单层时间轮（HashedWheelTimer）

任务数 > 10 万 且 delay 跨度大（ms ~ 小时混用）
    → 层级时间轮（Kafka TimingWheel）

参考资料

1. JDK 源码 — java.util.concurrent.ScheduledThreadPoolExecutor，DelayedWorkQueue，ScheduledFutureTask（OpenJDK GitHub）
2. Varghese & Lauck (1987) — Hashed and Hierarchical Timing Wheels: Data Structures for the Efficient Implementation of a Timer Facility — 时间轮原始论文，SOSP '87（对比最小堆的背景章节）
3. 《Java 并发编程实战》 — Brian Goetz 等，第 6 章任务执行，ScheduledExecutorService 使用规范
4. Doug Lea — A Java Fork/Join Framework — 线程池设计思想，leader-follower 模式出处
5. Linux Kernel — include/linux/timer.h，Linux 2.4 早期定时器基于最小堆，2.6 后改为层级时间轮
6. 《算法导论》第 4 版 — CLRS，第 6 章 Heapsort，堆的 siftUp / siftDown 标准实现