Go back

ScheduledThreadPoolExecutor：定时任务调度的正确打开方式

一句话结论（30s）

ScheduledThreadPoolExecutor 的本质是”多线程线程池 + 基于最小堆的 DelayedWorkQueue”，因为 Timer 的单线程会把慢任务阻塞、异常会静默杀死整条调度线。关键设计是 DelayedWorkQueue 用数组实现二叉堆，堆顶永远是最近要执行的任务，取堆顶 O(1)、插入删除 O(log n)。权衡是牺牲 Timer 的简单，换来多线程隔离、抗 NTP 漂移（nanoTime）和异常后继续服务的可靠性。

核心原理（2min）

任务先进入 DelayedWorkQueue，worker 线程 take 时看堆顶：若 delay<=0 取走并下沉堆化，否则 awaitNanos(delay) 等待；到点后交给线程池并行执行。调度分两种模式——fixed-rate 用”上一次计划开始时间 + period”追节奏（适合监控采样），fixed-delay 用”上一次实际完成时间 + delay”（适合批处理避免堆积）。

底层深入（5-10min）

Java的定时任务调度经历了从Timer到ScheduledThreadPoolExecutor的演进。如果你还在用Timer，可能需要重新审视——它有一些你未必意识到的致命缺陷。

Timer的问题：单线程的定时炸弹

Timer的实现非常直接：一个后台线程 + 一个优先级队列（TaskQueue，内部是二叉堆的实现）。后台线程循环从队列中取任务，计算等待时间，sleep到点执行。

问题就出在”一个后台线程”上：

问题一：任务互相阻塞。 如果前一个任务执行时间超过了它的周期，后一个任务就得排队等。所有TimerTask共享一个线程，排期再精确也没用。

想一想：为什么单线程的 Timer 排期再精确也没用？因为所有 TimerTask 共用一个后台线程，前一个慢任务占着线程，后一个到点也只能排队等——单线程决定了”排期精确”无法兑现为”按时执行”。

问题二：异常静默失败。 Timer线程如果抛出未捕获的异常，这个线程就挂了——而且不会重启。之后所有排好的任务都不会再执行，而且没有任何日志和警告。

想一想：为什么一个异常会”杀死”整条调度线？因为 Timer 的后台线程抛出未捕获异常就终止、且不会重启，后续所有已排任务都跟着失联，还不会有任何日志——这是单线程模型最隐蔽的坑。

Timer timer = new Timer();
timer.schedule(new TimerTask() {
    public void run() { throw new RuntimeException("炸了"); }
}, 1000);
timer.schedule(new TimerTask() {
    public void run() { System.out.println("永远不会打印"); }
}, 2000);
// 第一个任务异常后，第二个任务永远不会执行

问题三：时间漂移。 fixed-delay模式下（schedule方法），Timer用System.currentTimeMillis()做时间基准，系统时间被NTP调整会导致任务提前或延迟执行。

ScheduledThreadPoolExecutor的架构

ScheduledThreadPoolExecutor extends ThreadPoolExecutor，本质是多线程线程池 + 专用延迟队列。

核心数据结构是DelayedWorkQueue——一个基于**最小堆（二叉堆）**的无界阻塞队列。堆顶始终是最近需要执行的任务。

DelayedWorkQueue的内部实现

下面是 JDK 中 DelayedWorkQueue 的真实源码（核心字段 + offer/take + 堆化）：

// 核心字段
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();

// 插入任务（offer）：尾部插入 + 上浮
public boolean offer(Runnable x) {
    if (x == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int i = size;
        if (i >= queue.length)
            grow();
        size = i + 1;
        if (i == 0) {
            queue[0] = e;
            setIndex(e, 0);
        } else {
            siftUp(i, e);
        }
        if (queue[0] == e) {
            leader = null;
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
}

// 取任务（take）：看堆顶，没到点就等
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();
    }
}

// 上浮：与父节点比较，时间早的上浮
private void siftUp(int k, RunnableScheduledFuture<?> key) {
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        RunnableScheduledFuture<?> e = queue[parent];
        if (key.compareTo(e) >= 0)
            break;
        queue[k] = e;
        setIndex(e, k);
        k = parent;
    }
    queue[k] = key;
    setIndex(key, k);
}

// 下沉：与较小的子节点比较，时间晚的下沉
private void siftDown(int k, RunnableScheduledFuture<?> key) {
    int half = size >>> 1;
    while (k < half) {
        int child = (k << 1) + 1;
        RunnableScheduledFuture<?> c = queue[child];
        int right = child + 1;
        if (right < size && c.compareTo(queue[right]) > 0)
            c = queue[child = right];
        if (key.compareTo(c) <= 0)
            break;
        queue[k] = c;
        setIndex(c, k);
        k = child;
    }
    queue[k] = key;
    setIndex(key, k);
}

// 取堆顶：末尾元素补到堆顶，再下沉堆化
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
    int s = --size;
    RunnableScheduledFuture<?> x = queue[s];
    queue[s] = null;
    if (s != 0)
        siftDown(0, x);
    setIndex(f, -1);
    return f;
}

插入/删除的时间复杂度是O(log n)，取堆顶是O(1)。用数组存堆而非链表的对象节点，既有缓存友好性（数据连续）又避免了指针开销。take() 里藏着一个 Leader-Follower 模式：只有一个”leader”线程用 awaitNanos(delay) 精确等到队头任务到点，其余线程 available.await() 无限等，leader 醒来后 signal() 唤醒下一个——避免所有线程一起醒、一起抢、又一起睡回去的”惊群”。

想一想：为什么只让一个 leader 线程 awaitNanos 精确等待？因为多个线程一起精确等待会”惊群”——一起醒、一起抢、又一起睡回去，白白浪费 CPU；用 leader 独占等待、醒来 signal 下一个，就能有序传递。

拿到任务后，ScheduledThreadPoolExecutor 通过 delayedExecute() 把它塞进队列并确保有线程消费：

private void delayedExecute(RunnableScheduledFuture<?> task) {
    if (isShutdown())
        reject(task);
    else {
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(task) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

ensurePrestart() 会在 workerCount 为 0 时补一个 addWorker(null, false) 线程去消费队列——这正是上一篇文章里 execute() 二次检查的同一套兜底逻辑。入队后若池子刚好 shutdown，则把任务 remove 掉并 cancel，避免定时任务悬空。不同的定时任务可以被不同的线程并行执行，互不阻塞。

fixed-rate vs fixed-delay：两种调度模式的本质区别

JDK 里用一个 period 字段的正负号区分两种模式——正数 fixed-rate，负数 fixed-delay，0 表示一次性任务：

/**
 * Period for repeating tasks, in nanoseconds.
 * A positive value indicates fixed-rate execution.
 * A negative value indicates fixed-delay execution.
 * A value of 0 indicates a non-repeating (one-shot) task.
 */
private final long period;
// 计算下一次执行时间
private void setNextRunTime() {
    long p = period;
    if (p > 0)
        time += p;              // fixed-rate：上次计划时间 + period
    else
        time = triggerTime(-p); // fixed-delay：now + delay
}
// ScheduledFutureTask.run()：周期任务执行后重新入队
public void run() {
    if (!canRunInCurrentRunState(this))
        cancel(false);
    else if (!isPeriodic())
        super.run();
    else if (super.runAndReset()) {
        setNextRunTime();
        reExecutePeriodic(outerTask);
    }
}

而提交入口里，scheduleAtFixedRate 传正 period、scheduleWithFixedDelay 传负 delay：

// scheduleAtFixedRate：period 为正
ScheduledFutureTask<Void> sft =
    new ScheduledFutureTask<Void>(command,
                                  null,
                                  triggerTime(initialDelay, unit),
                                  unit.toNanos(period),
                                  sequencer.getAndIncrement());

// scheduleWithFixedDelay：delay 取负
ScheduledFutureTask<Void> sft =
    new ScheduledFutureTask<Void>(command,
                                  null,
                                  triggerTime(initialDelay, unit),
                                  -unit.toNanos(delay),
                                  sequencer.getAndIncrement());

scheduleAtFixedRate（固定速率）：

|--任务1--|  |--任务2--|  |--任务3--|
0s         10s        20s        30s

下一次任务的开始时间 = 上一次任务的计划开始时间 + period。即使上一次任务执行了很长时间，下一次也会尽量追回节奏——如果上一次延迟了太久导致多次执行”积压”，会合并执行。

想一想：为什么 fixed-rate 要”追节奏”而不是”顺延”？因为它以”计划开始时间 + period”为基准，即使上次执行慢了，下一次也按固定频率补，必要时把积压的多次执行合并——保证采集/心跳有稳定的时间节拍。

scheduleWithFixedDelay（固定延迟）：

|--任务1--|  |  |--任务2--|  |  |--任务3--|
0s      10s  15s        25s  28s        38s

下一次任务的开始时间 = 上一次任务的实际完成时间 + delay。前一个任务结束了才开始计时，不受执行时长影响。

这两种模式的选择取决于业务语义：

// fixed-rate：每10秒发一次心跳
scheduler.scheduleAtFixedRate(() -> sendHeartbeat(), 0, 10, TimeUnit.SECONDS);

// fixed-delay：上一批处理完后等30秒再跑下一批
scheduler.scheduleWithFixedDelay(() -> processBatch(), 0, 30, TimeUnit.SECONDS);

ScheduledThreadPoolExecutor vs Timer 对比总结

特性TimerScheduledThreadPoolExecutor
线程模型单线程多线程池
异常处理线程死亡，任务丢失任务异常后线程继续服务
任务隔离无，互相阻塞不同任务不同线程，隔离
调度精度依赖System.currentTimeMillis依赖System.nanoTime，抗NTP
追赶策略无fixed-rate可追赶

使用建议

章末提问

追问 1：为什么生产环境不该用 Timer？

回答思路：结论先行——Timer 是单线程，有三个致命缺陷：任务互相阻塞、异常静默杀死整条调度线、用 currentTimeMillis 受 NTP 漂移影响。因为所有 TimerTask 共享一个后台线程，前一个慢任务阻塞后一个；线程抛未捕获异常就终止且不重启、后续任务全部丢失；时间基准用墙钟时间，NTP 调时会导致任务提前或延迟。

追问 2：DelayedWorkQueue 凭什么能做到取堆顶 O(1)、插入 O(log n)？

回答思路：结论先行——因为它用数组实现最小二叉堆，堆顶永远是最近要执行的任务。因为堆顶取走时把末尾元素补到根再下沉堆化（siftDown），插入时尾插上浮（siftUp），都是 O(log n)；数组存堆比链表节点更缓存友好、省指针开销。

追问 3：fixed-rate 和 fixed-delay 的本质区别？怎么选？

回答思路：结论先行——fixed-rate 用”上一次计划开始时间 + period”追节奏，fixed-delay 用”上一次实际完成时间 + delay”。因为前者即使上次执行慢也按固定频率补、必要时合并积压任务，适合监控采样；后者等上次完成后才计时、避免堆积，适合批处理。

Share this post on:
Share this post via WhatsAppShare this post on FacebookShare this post on XShare this post via TelegramShare this post on PinterestShare this post via email
Previous Post
ThreadLocal内存泄漏的完整链路
Next Post
MDC的ThreadLocal泄漏与ScopedValue