跳转到内容

JUC线程池-ScheduledThreadPoolExecutor

草稿难度:中级#JUC

原文:CSDN(历史文章导入,当前状态为草稿)

1. 引言:为什么需要 ScheduledThreadPoolExecutor?

Section titled “1. 引言:为什么需要 ScheduledThreadPoolExecutor?”

在 Java 开发中,我们经常会遇到需要延迟执行或周期性执行任务的场景。比如:

  • 定时报表生成:每天凌晨生成前一天的业务报表。
  • 系统监控:每隔 5 分钟检查一次服务器的健康状态。
  • 缓存清理:每小时清理一次过期的缓存数据。
  • 任务重试:网络请求失败后,延迟 1 秒后重试。
  • 心跳检测:客户端与服务器之间定时发送心跳包维持连接。

对于这些需求,Java 提供了多种解决方案。早期开发者可能会想到 java.util.Timerjava.util.TimerTask。然而,Timer 存在一些固有的缺陷,使得它在现代并发应用中不再是首选:

  1. 单线程执行Timer 内部仅使用单个后台线程来执行所有任务。如果某个任务执行时间过长,会阻塞后续任务的执行。更糟糕的是,如果某个 TimerTask 抛出了未捕获的异常,这个唯一的执行线程会终止,导致 Timer 实例中的所有后续任务都无法再执行。
  2. 对系统时间敏感Timer 的调度是基于系统绝对时间的。如果系统时间发生改变(例如,用户手动修改时间),可能会导致任务执行混乱。
  3. 异常处理不佳:如上所述,未捕获的异常会导致整个 Timer 失效,这在健壮性要求高的系统中是不可接受的。

为了克服 Timer 的这些缺点,java.util.concurrent 包(JUC)中引入了 ScheduledThreadPoolExecutor。它是一个功能强大且灵活的线程池实现,专门用于处理定时和周期性任务。

ScheduledThreadPoolExecutor 的核心优势:

  • 多线程执行:它是一个真正的线程池,可以使用多个工作线程并发执行任务,避免了单线程瓶颈。
  • 相对时间调度:主要基于相对时间进行调度,不易受系统绝对时间变化的影响。
  • 健壮的异常处理:单个任务抛出异常不会影响其他任务的执行,也不会导致执行线程终止。线程池内部会捕获异常,并可以选择性地处理它们。
  • 更灵活的调度:提供了 scheduleAtFixedRate (固定速率) 和 scheduleWithFixedDelay (固定延迟) 两种周期性调度策略,满足不同场景的需求。
  • 继承自 ThreadPoolExecutor:拥有 ThreadPoolExecutor 的所有优点,如线程管理、拒绝策略、生命周期控制等。

简单来说,ScheduledThreadPoolExecutor 就像一个装备精良、训练有素的“定时任务特种部队”,相比于“单兵作战”的 Timer,它更强大、更可靠、更能适应复杂的并发环境。

本教程将带你深入探索 ScheduledThreadPoolExecutor 的内部世界,从基本用法到核心原理,再到源码分析和最佳实践,助你完全掌握这个强大的并发工具。

在深入源码之前,我们需要理解 ScheduledThreadPoolExecutor 的几个核心概念和基本使用方法。

2.1. ScheduledThreadPoolExecutor 与 ThreadPoolExecutor 的关系

Section titled “2.1. ScheduledThreadPoolExecutor 与 ThreadPoolExecutor 的关系”

ScheduledThreadPoolExecutor 继承自 ThreadPoolExecutor,并实现了 ScheduledExecutorService 接口。

public class ScheduledThreadPoolExecutor
extends ThreadPoolExecutor
implements ScheduledExecutorService {
// ...
}
  • 继承 ThreadPoolExecutor:这意味着 ScheduledThreadPoolExecutor 本质上是一个线程池,拥有线程池的核心能力,包括核心线程数(corePoolSize)、最大线程数(maximumPoolSize)、线程存活时间(keepAliveTime)、工作队列(workQueue)、线程工厂(threadFactory)和拒绝策略(handler)等概念。你可以像配置普通 ThreadPoolExecutor 一样配置这些参数(尽管某些参数对于 ScheduledThreadPoolExecutor 有特殊的默认行为或含义)。
  • 实现 ScheduledExecutorService:这个接口定义了执行延迟任务和周期性任务的标准方法,如 schedule()scheduleAtFixedRate()scheduleWithFixedDelay()ScheduledThreadPoolExecutor 提供了这些方法的具体实现。

可以理解为:ScheduledThreadPoolExecutor = ThreadPoolExecutor (线程池管理) + ScheduledExecutorService (定时调度能力)。

它复用了 ThreadPoolExecutor 的线程管理和任务执行框架,但专门针对定时任务的需求进行了定制和扩展。

2.2. 关键组件:DelayedWorkQueue 和 ScheduledFutureTask

Section titled “2.2. 关键组件:DelayedWorkQueue 和 ScheduledFutureTask”

ScheduledThreadPoolExecutor 的定时调度魔法主要依赖于两个关键组件:

  1. DelayedWorkQueue (延迟工作队列)

    • 这是一个无界(理论上容量是 Integer.MAX_VALUE)的阻塞队列,专门用于存放 ScheduledFutureTask
    • 它基于优先级堆(通常是最小堆)实现。队列中的任务并非按照先进先出(FIFO)的顺序排列,而是按照任务的下次执行时间排序,最近需要执行的任务排在队首
    • 只有当任务的延迟时间到了,它才能从队列中被取出 (take())。如果队首任务的执行时间还未到,尝试获取任务的线程会被阻塞,直到时间到达或被中断。
    • 这种设计使得工作线程可以高效地获取即将到期的任务,而无需轮询检查,大大降低了 CPU 消耗。
  2. ScheduledFutureTask (可调度的 Future 任务)

    • 当你通过 schedule() 等方法提交一个 RunnableCallable 时,ScheduledThreadPoolExecutor 会将其包装成一个 ScheduledFutureTask 对象。
    • ScheduledFutureTask 继承自 FutureTask(提供了获取异步执行结果、取消任务等能力),并实现了 RunnableScheduledFuture 接口。
    • RunnableScheduledFuture 接口又继承了 RunnableFutureDelayed 接口。
      • Delayed 接口要求实现 getDelay(TimeUnit unit) 方法,用于返回任务剩余的延迟时间,这是 DelayedWorkQueue 能够进行时间排序的关键。
      • RunnableScheduledFuture 接口还定义了 isPeriodic() 方法,用于判断任务是否是周期性的。
    • ScheduledFutureTask 内部维护了任务的下次执行时间 (time)、任务的类型(单次、固定速率、固定延迟)以及周期 (period) 等信息。
    • 它实现了 compareTo() 方法,允许 DelayedWorkQueue 根据执行时间对其进行排序。如果执行时间相同,则根据提交顺序(一个递增的 sequenceNumber)排序,保证 FIFO。
    • 对于周期性任务,ScheduledFutureTaskrun() 方法在执行完任务逻辑后,会重新计算下次执行时间,并将自身重新添加到 DelayedWorkQueue,从而实现周期性调度。

工作流程简述:

  1. 用户调用 schedule()scheduleAtFixedRate()scheduleWithFixedDelay() 提交任务。
  2. ScheduledThreadPoolExecutor 将任务包装成 ScheduledFutureTask,计算出首次执行时间。
  3. ScheduledFutureTask 被添加到 DelayedWorkQueue 中。队列内部根据执行时间(time)使用堆结构进行排序。
  4. 线程池中的工作线程循环调用 DelayedWorkQueuetake() 方法尝试获取任务。
  5. 如果队首任务未到期,take() 方法会阻塞工作线程。
  6. 当队首任务到期时,take() 方法返回该 ScheduledFutureTask
  7. 工作线程执行 ScheduledFutureTaskrun() 方法。
  8. 如果任务是周期性的,run() 方法内部会计算下次执行时间,并重新将该 ScheduledFutureTask 添加回 DelayedWorkQueue

这个流程巧妙地结合了线程池、优先级队列和特殊任务封装,实现了高效、可靠的定时调度。

ScheduledExecutorService 接口定义了三种主要的调度方法,ScheduledThreadPoolExecutor 提供了它们的实现:

  1. schedule(Runnable command, long delay, TimeUnit unit) / schedule(Callable<V> callable, long delay, TimeUnit unit)

    • 作用:安排一个任务在指定的 delay 之后执行一次
    • 参数
      • command/callable: 要执行的任务。
      • delay: 延迟执行的时间。
      • unit: delay 的时间单位(如 TimeUnit.SECONDS, TimeUnit.MILLISECONDS)。
    • 返回:一个 ScheduledFuture,可以用来取消任务或获取结果(对于 Callable)。
    • 场景:适用于只需要延迟执行一次的场景,例如:用户操作后延迟 1 秒显示提示信息,或者一个短暂的资源锁定,延迟释放。
  2. scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

    • 作用:安排一个任务在指定的 initialDelay 之后首次执行,然后以固定的速率 (fixed rate) 重复执行。
    • 参数
      • command: 要执行的任务。
      • initialDelay: 首次执行的延迟时间。
      • period: 两次执行开始时间之间的间隔。
      • unit: initialDelayperiod 的时间单位。
    • 返回:一个 ScheduledFuture,可以用来取消任务。注意:获取结果通常无意义,因为任务会持续执行。
    • 执行时间点计算
      • 首次执行时间点 = 当前时间 + initialDelay
      • 第二次执行开始时间点 = 首次执行开始时间点 + period
      • 第三次执行开始时间点 = 首次执行开始时间点 + 2 * period
      • 第 N 次执行开始时间点 = 首次执行开始时间点 + (N-1) * period
    • 关键特点:执行时间的计算是基于初始延迟时间点的,与每次任务实际执行花费的时间无关。如果某次任务执行时间超过了 period,那么下次任务会在前一次任务结束后立即开始,试图“追赶”上预定的时间点。这可能导致任务执行的并发(如果线程池大小 > 1)或者连续执行(没有间隔)。
    • 场景:适用于需要严格按照固定频率触发的场景,不管任务执行多久。例如:每秒钟刷新一次股票价格,每分钟记录一次系统性能指标。
  3. scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

    • 作用:安排一个任务在指定的 initialDelay 之后首次执行,然后在每次执行结束之后,再延迟一个固定的时间 (delay) 后重复执行。
    • 参数
      • command: 要执行的任务。
      • initialDelay: 首次执行的延迟时间。
      • delay: 上一次执行结束到下一次执行开始之间的延迟。
      • unit: initialDelaydelay 的时间单位。
    • 返回:一个 ScheduledFuture,可以用来取消任务。
    • 执行时间点计算
      • 首次执行时间点 = 当前时间 + initialDelay
      • 第二次执行开始时间点 = 第一次执行完成时间点 + delay
      • 第三次执行开始时间点 = 第二次执行完成时间点 + delay
    • 关键特点:执行时间的计算是基于上一次任务的完成时间点的。它确保了两次任务执行之间至少delay 的间隔。无论任务执行多长时间,执行间隔是固定的。
    • 场景:适用于需要确保两次执行之间有明确间隔的场景,或者任务执行时间不固定,不希望因执行时间过长导致任务堆积。例如:轮询检查某个资源是否可用,每次检查后等待 5 秒再进行下一次检查;或者失败任务的重试,每次重试间隔固定时间。

选择困难?一张图帮你理解 FixedRate vs FixedDelay:

假设 perioddelay 都设置为 3 秒,任务执行需要 1 秒。

FixedRate (period=3s):
任务1开始 --- 执行1s --> 任务1结束 |--- 等待2s ---| 任务2开始 --- 执行1s --> 任务2结束 |--- 等待2s ---| 任务3开始 ...
|----------------- 3s -----------------|----------------- 3s -----------------|
FixedDelay (delay=3s):
任务1开始 --- 执行1s --> 任务1结束 |---------- 等待3s ----------| 任务2开始 --- 执行1s --> 任务2结束 |---------- 等待3s ----------| 任务3开始 ...
|-------------------- 4s --------------------|-------------------- 4s --------------------|

假设 perioddelay 都设置为 3 秒,任务执行需要 4 秒。

FixedRate (period=3s):
任务1开始 ----------- 执行4s -----------> 任务1结束 | 任务2开始 ----------- 执行4s -----------> 任务2结束 | 任务3开始 ...
|----------------- 3s (理论) ---------------| (立即开始)
|----------------- 3s (理论) ---------------| (立即开始)
^
注意:下次任务紧接着开始,尝试追赶
FixedDelay (delay=3s):
任务1开始 ----------- 执行4s -----------> 任务1结束 |---------- 等待3s ----------| 任务2开始 ----------- 执行4s -----------> 任务2结束 |---------- 等待3s ----------| 任务3开始 ...
|------------------------- 7s -------------------------|------------------------- 7s -------------------------|

总结:

  • 需要固定频率,不关心执行时间是否重叠 -> scheduleAtFixedRate
  • 需要固定间隔,确保执行之间有喘息 -> scheduleWithFixedDelay
  • 只需要执行一次 -> schedule
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class ScheduledThreadPoolDemo {
public static void main(String[] args) throws InterruptedException {
// 创建一个核心线程数为 2 的 ScheduledThreadPoolExecutor
// 使用默认的线程工厂和拒绝策略 (AbortPolicy)
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
System.out.println("系统时间: " + System.currentTimeMillis());
// 1. 延迟 3 秒后执行一次的任务 (Runnable)
scheduler.schedule(() -> {
System.out.println("延迟任务执行 @ " + System.currentTimeMillis() + " by " + Thread.currentThread().getName());
}, 3, TimeUnit.SECONDS);
// 2. 延迟 1 秒后首次执行,然后每隔 2 秒执行一次 (固定速率)
// 注意:如果任务执行时间超过 2 秒,下次任务会在上次结束后立即执行
AtomicInteger rateCounter = new AtomicInteger(0);
ScheduledFuture<?> rateHandle = scheduler.scheduleAtFixedRate(() -> {
int count = rateCounter.incrementAndGet();
long start = System.currentTimeMillis();
System.out.println("Rate任务 " + count + " 开始 @ " + start + " by " + Thread.currentThread().getName());
try {
// 模拟任务执行耗时
TimeUnit.MILLISECONDS.sleep(1000); // 任务执行 1 秒
// TimeUnit.MILLISECONDS.sleep(2500); // 任务执行 2.5 秒 (测试执行时间超过 period 的情况)
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Rate任务 " + count + " 结束 @ " + System.currentTimeMillis() + ", 耗时: " + (System.currentTimeMillis() - start) + "ms");
}, 1, 2, TimeUnit.SECONDS);
// 3. 延迟 1 秒后首次执行,然后在上次任务结束后延迟 2 秒再执行 (固定延迟)
AtomicInteger delayCounter = new AtomicInteger(0);
scheduler.scheduleWithFixedDelay(() -> {
int count = delayCounter.incrementAndGet();
long start = System.currentTimeMillis();
System.out.println("Delay任务 " + count + " 开始 @ " + start + " by " + Thread.currentThread().getName());
try {
// 模拟任务执行耗时
TimeUnit.MILLISECONDS.sleep(1000); // 任务执行 1 秒
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Delay任务 " + count + " 结束 @ " + System.currentTimeMillis() + ", 耗时: " + (System.currentTimeMillis() - start) + "ms");
}, 1, 2, TimeUnit.SECONDS);
// 让主线程等待一段时间,以便观察定时任务的执行
// TimeUnit.SECONDS.sleep(10);
// 可以在一段时间后取消某个周期任务
// TimeUnit.SECONDS.sleep(7);
// System.out.println("尝试取消 Rate 任务...");
// rateHandle.cancel(false); // false: 不中断正在执行的任务; true: 中断正在执行的任务
// System.out.println("Rate 任务是否已完成 (或取消): " + rateHandle.isDone());
// 优雅关闭线程池
// scheduler.shutdown(); // 不再接受新任务,等待已提交任务执行完成
// try {
// // 等待最多 10 秒让任务执行完毕
// if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
// System.err.println("线程池任务未在指定时间内完成,强制关闭...");
// scheduler.shutdownNow(); // 尝试取消所有正在执行和等待的任务
// }
// } catch (InterruptedException e) {
// System.err.println("等待线程池关闭时被中断,强制关闭...");
// scheduler.shutdownNow();
// Thread.currentThread().interrupt();
// }
// System.out.println("线程池已关闭.");
// 注意:为了方便观察,示例中没有关闭线程池,实际应用中必须关闭!
// 如果不关闭,主线程结束了,但线程池的线程(非守护线程)会阻止 JVM 退出。
}
}

你可以尝试运行上面的代码,并修改 scheduleAtFixedRate 中的任务执行时间(注释掉 1 秒的 sleep,打开 2.5 秒的 sleep),观察 FixedRateFixedDelay 在任务执行时间超过 period/delay 时的不同表现。

ScheduledThreadPoolExecutor 在异常处理方面比 Timer 健壮得多:

  • 工作线程受保护ThreadPoolExecutorScheduledThreadPoolExecutor 的父类)在执行任务的 runWorker 方法内部有一个大的 try-finally 块。即使提交的任务本身没有捕获异常,这个 try-finally 块也会捕获 Throwable,确保工作线程不会因为任务抛出的异常而死亡。工作线程在处理完一个(可能异常退出的)任务后,会继续从队列中获取下一个任务执行。
  • 周期任务的特殊处理
    • 如果通过 scheduleAtFixedRatescheduleWithFixedDelay 提交的周期性任务在某次执行时抛出了未捕获的异常,那么该任务的后续执行将被取消。它不会再被重新调度。
    • 对于单次任务(通过 schedule 提交),抛出异常只会导致该次任务执行失败,其 Future.get() 会抛出 ExecutionException
  • 异常默认被“吞掉”:默认情况下,工作线程捕获到的异常(源自任务内部)不会被打印到控制台,也不会被重新抛出。这可能导致问题被悄无声息地忽略。
  • 如何捕获和处理异常
    1. 在任务内部 try-catch:最直接的方式是在你的 RunnableCallablerun/call 方法内部添加 try-catch 块,自行处理异常(如记录日志)。这是推荐的做法。

    2. 重写 afterExecute 方法:可以继承 ScheduledThreadPoolExecutor 并重写 protected void afterExecute(Runnable r, Throwable t) 方法。这个方法在每个任务执行完毕后被调用。如果任务正常完成,tnull;如果任务抛出异常,t 就是那个异常对象。你可以在这里统一处理异常,比如记录日志。注意,需要检查 r 是否是 Future<?> 类型,并调用其 get() 方法来触发潜在的 ExecutionException,然后处理 t 或捕获的 ExecutionException

      ScheduledThreadPoolExecutor customScheduler = new ScheduledThreadPoolExecutor(1) {
      @Override
      protected void afterExecute(Runnable r, Throwable t) {
      super.afterExecute(r, t);
      if (t == null && r instanceof Future<?>) {
      try {
      Future<?> future = (Future<?>) r;
      if (future.isDone()) {
      future.get(); // 如果任务内部有异常,这里会抛出 ExecutionException
      }
      } catch (CancellationException ce) {
      // 任务被取消,通常无需处理
      t = ce;
      } catch (ExecutionException ee) {
      // 任务执行时抛出异常
      t = ee.getCause(); // 获取原始异常
      } catch (InterruptedException ie) {
      // 当前线程被中断
      Thread.currentThread().interrupt(); // Preserve interrupt status
      }
      }
      if (t != null) {
      System.err.println("任务执行异常: " + t);
      // 在这里添加日志记录逻辑
      // logger.error("Scheduled task threw an exception", t);
      }
      }
      };
      + 1
      + 2
      + 3
      + 4
      + 5
      + 6
      + 7
      + 8
      + 9
      + 10
      + 11
      + 12
      + 13
      + 14
      + 15
      + 16
      + 17
      + 18
      + 19
      + 20
      + 21
      + 22
      + 23
      + 24
      + 25
      + 26
      + 27
      + 28
    3. 使用 ThreadFactory 设置 UncaughtExceptionHandler:创建一个自定义的 ThreadFactory,为线程池创建的每个线程设置一个 UncaughtExceptionHandler。当线程因未捕获的异常而即将终止时(虽然 ThreadPoolExecutor 的工作线程通常不会因此终止,但设置处理器总是一个好习惯,且可能捕获其他意外情况),这个处理器会被调用。但这对于处理 任务内部 的异常不是主要方式,因为 ThreadPoolExecutor 通常会捕获它们。

强烈建议在任务代码内部处理可预见的异常,并通过 afterExecutetry-catch Future.get() 来捕获和记录未预见的异常。

理解了基本概念后,让我们深入 ScheduledThreadPoolExecutor 的源码(基于 OpenJDK 11,但核心逻辑在多个版本中类似),看看它是如何精确地实现定时调度的。

ScheduledThreadPoolExecutor 提供了多个构造函数,最常用的是:

public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}

关键点分析:

  1. maximumPoolSize 固定为 Integer.MAX_VALUE:这意味着 ScheduledThreadPoolExecutor 在理论上可以创建无限多的线程(受系统资源限制)。然而,这通常不是问题,因为任务是由 DelayedWorkQueue 控制何时执行的。只有当大量任务同时到期时,才可能创建超过 corePoolSize 的线程。但在实践中,corePoolSize 通常是主要的性能瓶颈。
  2. keepAliveTime 固定为 0:这意味着当线程池中的线程数超过 corePoolSize 时,多余的空闲线程会立即被终止。但由于 maximumPoolSize 极大,这个参数在 ScheduledThreadPoolExecutor 的默认行为下几乎不起作用,除非你自己用其他构造函数或 setMaximumPoolSize() 强制设置了一个较小的 maximumPoolSize
  3. workQueue 固定为 DelayedWorkQueue:这是实现定时调度的核心队列,后面会详细分析。
  4. corePoolSize:这是唯一必须指定的参数,表示线程池中保持活动状态的最小线程数。即使它们是空闲的,也不会被回收(除非设置了 allowCoreThreadTimeOut(true),但默认是 false)。对于定时任务,保持核心线程活跃通常是必要的,以确保任务能够及时被执行。

思考:为什么 maximumPoolSize 设置为 Integer.MAX_VALUE

Doug Lea 在设计时可能考虑到,定时任务的执行时机是由任务本身的时间属性决定的,而不是由线程池的繁忙程度。如果大量任务恰好在同一时间点到期,使用有界的 maximumPoolSize 可能会导致任务无法及时执行,甚至被拒绝。将其设置为无界,并将调度的核心压力放在 DelayedWorkQueue 上,可以确保只要系统资源允许,到期的任务总能有线程去执行。同时,keepAliveTime 为 0 确保了这种“超额”创建的线程在空闲后能被快速回收。不过,这也意味着如果配置不当(如 corePoolSize 过小而任务并发度很高),可能会创建过多线程,消耗系统资源。因此,合理设置 corePoolSize 仍然非常重要。

我们来看 schedule(Callable<V> callable, long delay, TimeUnit unit) 的实现:

ScheduledThreadPoolExecutor.java
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
long delay,
TimeUnit unit) {
// 1. 参数校验:callable 和 unit 不能为空
if (callable == null || unit == null)
throw new NullPointerException();
// 2. 任务包装:将 Callable 包装成 ScheduledFutureTask
// decorateTask 方法默认直接返回传入的 task,允许子类进行扩展
RunnableScheduledFuture<V> t = decorateTask(callable,
new ScheduledFutureTask<V>(callable, triggerTime(delay, unit))); // 计算首次执行时间
// 3. 延迟执行:调用 delayedExecute 方法将任务放入队列
delayedExecute(t);
// 4. 返回 Future:返回包装后的 ScheduledFutureTask
return t;
}
// 计算任务的触发时间 (纳秒)
private long triggerTime(long delay, TimeUnit unit) {
// 将用户指定的延迟转换为纳秒
// 如果 delay 小于 0,则视为 0
return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
// 内部方法,根据纳秒延迟计算绝对触发时间
long triggerTime(long delay) {
// 使用 System.nanoTime() 获取当前纳秒时间,加上延迟
// nanoTime() 用于测量时间间隔,比 System.currentTimeMillis() 更精确,且不受系统时钟调整影响
// overflowFree 参数用于处理可能的 long 类型溢出,确保时间计算正确
return now() + overflowFree(delay);
}
// 获取当前纳秒时间
final long now() {
return System.nanoTime();
}
// 提交任务到队列的核心方法
private void delayedExecute(RunnableScheduledFuture<?> task) {
// 1. 检查线程池是否已关闭
if (isShutdown())
// 如果关闭,执行拒绝策略 (通常是抛出 RejectedExecutionException)
reject(task);
else {
// 2. 将任务添加到 DelayedWorkQueue 中
// 队列会根据 task.getDelay() (内部使用 time 字段) 进行排序
super.getQueue().add(task); // 注意调用的是父类 ThreadPoolExecutor 的 getQueue()
// 3. 再次检查线程池状态,并确保至少有一个工作线程在运行
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) && // 检查任务是否能在当前状态下运行
remove(task)) // 如果不能运行,尝试从队列中移除任务
task.cancel(false); // 移除成功,则取消任务
else
// 确保有工作线程启动,以处理队列中的任务
// 如果当前工作线程数小于 corePoolSize,会尝试启动一个新线程
// 如果工作线程数为 0,也会启动一个新线程
ensurePrestart(); // 这是 ThreadPoolExecutor 的方法
}
}

源码分析总结 (schedule):

  1. 参数检查:基本的非空检查。
  2. 计算触发时间 triggerTime:核心是使用 System.nanoTime() 获取当前纳秒时间,并加上用户指定的延迟(已转换为纳秒)。nanoTime() 用于精确测量时间间隔,是实现准确延迟的关键。使用 overflowFree 处理可能的 long 溢出。
  3. 任务包装 decorateTask & ScheduledFutureTask:将用户提交的 CallableRunnable 包装成 ScheduledFutureTask。这个 task 对象内部存储了计算好的触发时间 timedecorateTask 是一个钩子方法,允许子类在任务入队前对其进行修改或替换。
  4. 入队 delayedExecute
    • 先检查线程池是否已关闭,若关闭则执行拒绝策略。
    • 调用 super.getQueue().add(task) 将任务添加到 DelayedWorkQueue。队列内部会根据任务的 time 字段(通过 getDelay 方法体现)将其放置到优先级堆的正确位置。
    • 双重检查关闭状态:添加任务后,再次检查线程池是否关闭。这是一种必要的并发控制,防止在 isShutdown() 检查和 add() 操作之间线程池状态发生改变。如果线程池已关闭,并且该任务(特别是周期性任务)不允许在关闭状态下运行,则尝试从队列中移除该任务并取消它。
    • 确保工作线程存在 ensurePrestart:调用 ThreadPoolExecutorensurePrestart 方法。该方法确保至少有一个工作线程启动并运行。如果当前活动线程数小于 corePoolSize,或者线程池中没有任何线程,它会创建一个新的工作线程。这个工作线程启动后会立即尝试从 DelayedWorkQueuetake() 任务。如果队首任务未到期,线程会阻塞等待。

3.3. scheduleAtFixedRate 方法源码解析

Section titled “3.3. scheduleAtFixedRate 方法源码解析”
ScheduledThreadPoolExecutor.java
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
// 1. 参数校验
if (command == null || unit == null)
throw new NullPointerException();
if (period <= 0) // 周期必须是正数
throw new IllegalArgumentException();
// 2. 任务包装:
// - 将 Runnable 包装成 ScheduledFutureTask
// - 计算首次执行时间 triggerTime(initialDelay, unit)
// - 关键:传入 period 作为 ScheduledFutureTask 的 period 字段。
// 正的 period 表示这是一个 fixed-rate 任务。
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null, // Runnable 没有返回值
triggerTime(initialDelay, unit), // 首次执行时间
unit.toNanos(period)); // 周期 (纳秒)
// 3. 任务装饰 (同 schedule)
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
// 4. 记录任务的下次执行时间 (用于 shutdownNow 的处理)
sft.outerTask = t; // 让 sft 能够访问到最终放入队列的 task 对象 t
// 5. 延迟执行 (同 schedule)
delayedExecute(t);
// 6. 返回 Future
return t;
}
// ScheduledFutureTask.java 的构造函数 (部分)
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result); // 调用 FutureTask 构造函数
this.time = ns; // 存储首次执行时间 (纳秒)
this.period = period; // 存储周期 (纳秒),正数表示 fixed-rate
this.sequenceNumber = sequencer.getAndIncrement(); // 原子递增的序号,用于排序稳定性
}

源码分析总结 (scheduleAtFixedRate):

  1. 参数校验:增加了对 period 必须大于 0 的校验。
  2. 任务包装 ScheduledFutureTask:与 schedule 类似,但这次构造 ScheduledFutureTask 时传入了正的 period(转换为纳秒)。ScheduledFutureTask 内部会记录这个 period
  3. outerTask 引用sft.outerTask = t; 这行代码是为了让内部的 ScheduledFutureTask (sft) 能够持有最终被放入队列的(可能被 decorateTask 装饰过的)任务对象 t 的引用。这在 reExecutePeriodic(处理周期性任务重新入队)和 cancel 时可能用到。
  4. 入队 delayedExecute:流程与 schedule 完全相同。

关键在于 ScheduledFutureTask 内部如何处理这个正的 period。我们稍后在分析 ScheduledFutureTask.run() 时会看到。

3.4. scheduleWithFixedDelay 方法源码解析

Section titled “3.4. scheduleWithFixedDelay 方法源码解析”
ScheduledThreadPoolExecutor.java
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit) {
// 1. 参数校验
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0) // 延迟必须是正数
throw new IllegalArgumentException();
// 2. 任务包装:
// - 将 Runnable 包装成 ScheduledFutureTask
// - 计算首次执行时间 triggerTime(initialDelay, unit)
// - 关键:传入 -delay (负的延迟) 作为 ScheduledFutureTask 的 period 字段。
// 负的 period 表示这是一个 fixed-delay 任务。
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit), // 首次执行时间
unit.toNanos(-delay)); // 周期 (注意是负数!)
// 3. 任务装饰 (同上)
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
// 4. outerTask 引用 (同上)
sft.outerTask = t;
// 5. 延迟执行 (同上)
delayedExecute(t);
// 6. 返回 Future
return t;
}

源码分析总结 (scheduleWithFixedDelay):

代码结构与 scheduleAtFixedRate 几乎完全一样,唯一的、也是最关键的区别在于构造 ScheduledFutureTask 时传入的 period 参数:

  • scheduleAtFixedRate 传入的是 unit.toNanos(period) (正数)。
  • scheduleWithFixedDelay 传入的是 unit.toNanos(-delay) (负数)。

ScheduledFutureTask 内部通过检查 period 字段的正负来区分这两种周期性任务,并在计算下次执行时间时采用不同的逻辑。

3.5. ScheduledFutureTask.run() 源码解析:周期性调度的核心

Section titled “3.5. ScheduledFutureTask.run() 源码解析:周期性调度的核心”

ScheduledFutureTaskrun() 方法是实现周期性调度的关键所在。当工作线程从 DelayedWorkQueue 取出一个到期的 ScheduledFutureTask 并执行其 run() 方法时,会发生以下情况:

ScheduledFutureTask.java
public void run() {
// 1. 判断是否是周期性任务
boolean periodic = isPeriodic(); // 检查 period 是否非 0
// 2. 检查任务是否能在当前状态下执行 (例如,线程池未关闭)
if (!canRunInCurrentRunState(periodic)) {
cancel(false); // 不能运行则取消
return;
}
// 3. 如果是单次任务,直接调用 FutureTask.run() 执行
if (!periodic)
ScheduledFutureTask.super.run(); // 调用父类 FutureTask 的 run 方法
// 4. 如果是周期性任务,调用 FutureTask.runAndReset() 执行,并重新调度
else if (ScheduledFutureTask.super.runAndReset()) { // runAndReset 执行任务,成功则返回 true
// 5. 计算下次执行时间
setNextRunTime();
// 6. 重新将任务添加到队列
// 注意:这里调用的是 ScheduledThreadPoolExecutor 的 reExecutePeriodic 方法
// 该方法内部会再次调用 delayedExecute 将任务放回 DelayedWorkQueue
reExecutePeriodic(outerTask);
}
}
// 判断是否是周期任务
public boolean isPeriodic() {
return period != 0; // period 非 0 即为周期任务 (正为 fixed-rate, 负为 fixed-delay)
}
// 计算下次执行时间
private void setNextRunTime() {
long p = period; // 获取周期 (可能是正数或负数)
if (p > 0) { // 正数:Fixed-Rate 任务
// 下次执行时间 = 上次理论执行时间 + period
// 注意:这里是 time += p,是基于上一次的理论时间点累加,而不是基于当前完成时间
time += p;
} else { // 负数:Fixed-Delay 任务
// 下次执行时间 = 当前时间 + abs(period) (即设置的 delay)
// 使用 now() 获取任务完成后的当前时间
// 这里是 time = triggerTime(-p),-p 将负的 period 转为正的 delay
// triggerTime(delay) = now() + delay
time = triggerTime(-p);
}
}
// 重新执行周期任务 (由 ScheduledThreadPoolExecutor 提供)
// ScheduledThreadPoolExecutor.java
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
// 检查线程池是否允许执行 (未关闭)
if (canRunInCurrentRunState(true)) {
// 重新将任务添加到队列
super.getQueue().add(task);
// 再次检查关闭状态,如果期间关闭了,则移除并取消任务
if (!canRunInCurrentRunState(true) && remove(task))
task.cancel(false);
else
// 确保有工作线程,以防所有线程都在等待且新任务的唤醒丢失
ensurePrestart();
}
}
// FutureTask.java (父类方法)
// runAndReset() 大致逻辑:
// 1. 执行 Callable 或 Runnable 的 call/run 方法。
// 2. 如果执行成功,返回 true。
// 3. 如果执行过程中抛出异常,捕获异常,将 Future 状态设为异常完成,返回 false。
// 4. 与 run() 不同的是,runAndReset() 成功后不会将 Future 的状态设为完成 (outcome 不变),
// 允许任务被再次执行。但它不会保留执行结果。
protected boolean runAndReset() {
// ... 省略状态检查和 CAS 操作 ...
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
c.call(); // 执行任务
ran = true; // 标记已运行
}
} catch (Throwable ex) {
setException(ex); // 设置异常结果
return false; // 执行失败
}
// ... 省略后续状态处理 ...
return ran; // 返回执行是否成功
}

源码分析总结 (ScheduledFutureTask.run):

  1. 区分周期性与非周期性:通过 isPeriodic() (检查 period != 0) 判断任务类型。
  2. 单次任务 (period == 0):直接调用 FutureTask.super.run()。该方法会执行任务,设置结果(或异常),并将 Future 状态置为完成。任务执行一次后结束。
  3. 周期性任务 (period != 0)
    • 调用 FutureTask.super.runAndReset()。此方法会执行任务逻辑。如果任务成功执行(未抛异常),它会返回 true,并且不会改变 Future 的完成状态,也不会存储结果,允许任务被重新调度。如果任务抛出异常,runAndReset 会捕获异常,设置 Future 的异常结果,并返回 false
    • 如果 runAndReset 返回 true (任务成功执行)
      • 调用 setNextRunTime() 计算下一次执行的绝对纳秒时间 time
        • Fixed-Rate (period > 0): time = time + period (基于上次理论时间点)。
        • Fixed-Delay (period < 0): time = now() + abs(period) (基于当前完成时间点)。
      • 调用 ScheduledThreadPoolExecutorreExecutePeriodic() 方法,该方法内部本质上是再次调用 delayedExecute(),将更新了 time 字段的 ScheduledFutureTask 重新添加DelayedWorkQueue。队列会根据新的 time 值将其放到正确的位置。
    • 如果 runAndReset 返回 false (任务抛出异常):任务的 Future 状态已被设为异常完成,run() 方法结束,不会再调用 setNextRunTime()reExecutePeriodic()因此,周期性任务一旦抛出未捕获的异常,其后续调度就会终止。

这就是 ScheduledThreadPoolExecutor 实现周期性调度的核心机制:通过 ScheduledFutureTask 在每次成功执行后重新计算时间并重新入队,实现了任务的循环执行。并通过 period 的正负区分了 FixedRateFixedDelay 的不同计时逻辑。

3.6. DelayedWorkQueue 源码浅析:时间排序的奥秘

Section titled “3.6. DelayedWorkQueue 源码浅析:时间排序的奥秘”

DelayedWorkQueueScheduledThreadPoolExecutor 的心脏。它是一个基于最小堆实现的优先级队列,专门用于存储和管理 Delayed 对象(如 ScheduledFutureTask)。

// DelayedWorkQueue.java (内部结构简化)
static class DelayedWorkQueue extends AbstractQueue<Runnable>
implements BlockingQueue<Runnable> {
// 初始容量
private static final int INITIAL_CAPACITY = 16;
// 底层存储:数组实现的堆 (通常是最小堆)
private RunnableScheduledFuture<?>[] queue =
new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
// 用于队列并发控制的锁
private final ReentrantLock lock = new ReentrantLock();
// 当前队列中的元素数量
private int size = 0;
// 指向“领导者”线程,用于优化 take() 操作,减少不必要的 timedWait
private Thread leader = null;
// 条件变量,当队首任务到期或有新任务入队时,唤醒等待的线程
private final Condition available = lock.newCondition();
// ... (构造函数等) ...
// 入队操作
public boolean offer(Runnable x) {
// 不接受 null
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,将新任务按照优先级(执行时间)插入到堆的正确位置
siftUp(i, e);
}
// 如果新插入的任务成为了新的队首 (即最早执行的任务)
if (queue[0] == e) {
leader = null; // 清空 leader,让所有等待线程重新竞争
available.signal(); // 唤醒一个可能在 take() 中等待的线程
}
} finally {
lock.unlock(); // 释放锁
}
return true; // 无界队列,offer 总是返回 true
}
// 出队操作 (阻塞)
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly(); // 可中断地获取锁
try {
for (;;) { // 无限循环,直到获取到任务或被中断
RunnableScheduledFuture<?> first = queue[0]; // 查看队首任务
if (first == null)
// 队列为空,阻塞等待,直到有新任务入队 (available.signal())
available.await();
else {
// 队列不为空,获取队首任务的延迟时间
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
// 延迟时间已到或已过,说明任务可以执行
// 调用 finishPoll 将任务从堆中移除并返回
return finishPoll(first);
// 任务还未到期
first = null; // 释放 first 引用,避免内存泄漏
if (leader != null)
// 如果已有 leader 线程在等待,当前线程直接无限期等待
// 避免多个线程同时 timedWait 同一个任务,减少唤醒开销
available.await();
else {
// 没有 leader,当前线程成为 leader
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// timedWait: 等待指定的 delay 时间
// 如果期间有新任务插入且成为队首,或当前任务被取消,会被 signal 唤醒
available.awaitNanos(delay);
} finally {
// 等待结束 (超时或被唤醒),如果当前线程仍然是 leader,则清空
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒:如果 leader 为空且队列不为空,唤醒一个其他等待线程
// 确保即使 leader 拿到任务走了,也有其他线程能成为新的 leader
if (leader == null && queue[0] != null)
available.signal();
lock.unlock(); // 释放锁
}
}
// siftUp: 堆的上浮操作,用于插入新元素
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);
}
// siftDown: 堆的下沉操作,用于删除堆顶元素后调整
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);
}
// finishPoll: 完成 poll/take 操作,移除堆顶元素并返回
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
int s = --size; // 数量减 1
RunnableScheduledFuture<?> x = queue[s]; // 获取最后一个元素
queue[s] = null; // 帮助 GC
if (s != 0)
// 将最后一个元素放到堆顶,然后执行下沉操作,维持堆结构
siftDown(0, x);
setIndex(f, -1); // 标记移除的元素索引无效
return f;
}
// ... 其他方法如 peek, poll, remove, grow, setIndex 等 ...
}

源码分析总结 (DelayedWorkQueue):

  1. 数据结构:内部使用一个数组 queue 来实现最小堆。堆顶 queue[0] 始终是下次执行时间最早的那个 ScheduledFutureTask
  2. 排序依据siftUpsiftDown 操作依赖 ScheduledFutureTask 实现的 compareTo 方法进行比较。compareTo 首先比较 time (下次执行时间),时间早的优先级高;如果时间相同,则比较 sequenceNumber (任务提交顺序),序号小的优先级高。
  3. offer (入队):将新任务添加到数组末尾,然后调用 siftUp 将其“上浮”到堆中正确的位置。如果新任务成为新的堆顶 (queue[0]),则唤醒一个可能在 take() 中等待的线程。
  4. take (出队)
    • 检查队首 queue[0]
    • 如果队列为空,await() 无限期等待。
    • 如果队首任务已到期 (getDelay() <= 0),调用 finishPoll 将其从堆中移除(将最后一个元素移到堆顶再 siftDown 调整),然后返回该任务。
    • 如果队首任务未到期 (getDelay() > 0):
      • Leader/Follower 模式优化:引入 leader 线程。第一个发现任务未到期的线程成为 leader,并调用 awaitNanos(delay) 精确等待到任务的到期时间点。其他后续到达的线程(Follower)发现已有 leader,则直接 await() 无限期等待,避免多个线程对同一个未来时间点进行 timedWait,减少 CPU 消耗和不必要的唤醒。
      • leader 等待超时(任务到期)或被 signal() 唤醒(例如有更早的任务插入或任务被取消),它会重新进入循环检查队首。
      • 当一个线程(可能是原 leader 或其他线程)成功获取到任务并离开 take 方法时,它会在 finally 块中检查是否需要 signal() 唤醒另一个等待的线程,确保队列的处理能够持续进行。
  5. 并发控制:使用 ReentrantLockCondition (available) 来保证线程安全和高效的线程协作(等待与唤醒)。

DelayedWorkQueue 通过精巧的堆结构和优化的等待/唤醒机制,实现了让工作线程能够高效、低耗地获取到刚好到期的定时任务。

掌握了核心原理后,我们还需要了解一些高级主题和实践中的注意事项。

4.1. 合理设置线程池大小 (corePoolSize)

Section titled “4.1. 合理设置线程池大小 (corePoolSize)”

corePoolSizeScheduledThreadPoolExecutor 最重要的配置参数。设置过小可能导致任务延迟执行,设置过大则会浪费系统资源。如何确定合理的值?

需要考虑的因素:

  1. 任务的性质
    • CPU 密集型任务:任务需要大量的计算,很少等待 I/O。这种情况下,线程数不宜过多,通常设置为 CPU 核心数 + 1 是一个不错的起点。过多的线程会导致频繁的上下文切换,反而降低性能。
    • I/O 密集型任务:任务大部分时间在等待网络、磁盘或其他外部资源响应。例如,调用远程 API、数据库查询、文件读写。这种情况下,线程可以设置得多一些,因为线程在等待时不会占用 CPU。可以设置为 CPU 核心数 * (1 + 平均等待时间 / 平均计算时间),这是一个理论公式,实践中通常设置为 CPU 核心数 * 2 或更多,具体需要根据实际的 I/O 等待情况和系统负载来调整。
    • 混合型任务:包含计算和 I/O。需要根据任务中计算和 I/O 的比例来权衡。
  2. 任务的执行频率和并发度
    • 如果有很多任务需要同时到期并执行,corePoolSize 需要足够大以应对峰值并发。例如,如果预计最多有 10 个任务会同时到期执行,那么 corePoolSize 至少应设置为 10。
    • 如果任务执行非常频繁(例如,毫秒级的周期任务),需要确保有足够的线程来处理,避免任务堆积。
  3. 任务的执行时长
    • 短耗时任务:如果任务执行时间很短,即使并发量大,较小的 corePoolSize 可能也能应付。
    • 长耗时任务:如果任务执行时间很长,它们会长时间占用线程。如果同时有其他任务到期,就需要更多的线程来避免阻塞。对于非常耗时的周期任务,要特别小心 scheduleAtFixedRate 可能导致的执行堆积。
  4. 系统资源:服务器的 CPU、内存资源是硬性约束。线程数越多,内存占用(线程栈)和 CPU 上下文切换开销就越大。
  5. 依赖的外部系统:如果任务依赖于外部服务(如数据库、第三方 API),这些服务的并发处理能力也是一个限制因素。设置过高的线程数可能压垮下游服务。

实践建议:

  • 没有银弹公式:最佳线程数通常需要通过性能测试和监控来确定。
  • 从保守值开始:可以从一个较小的值(如 CPU 核心数)开始,然后根据应用的性能指标(任务延迟、队列长度、CPU 利用率、线程池活跃线程数等)逐步调整。
  • 使用监控工具:利用 JMX、Micrometer、Arthas 等工具监控线程池的各项指标(activeCount, poolSize, queueSize, taskCount, completedTaskCount),了解其实际运行状况。
  • 区分不同类型的任务:如果应用中有多种不同性质的定时任务(CPU密集 vs IO密集,长耗时 vs 短耗时),考虑使用多个 ScheduledThreadPoolExecutor 实例,为不同类型的任务配置不同的 corePoolSize,避免相互影响。
  • 考虑 allowCoreThreadTimeOut(true):如果你希望核心线程在空闲一段时间后也能被回收(适用于任务不频繁的场景,以节省资源),可以调用此方法并设置合适的 keepAliveTime。但请注意,这可能导致任务到来时需要重新创建线程,增加一点点延迟。对于需要快速响应的定时任务,通常不建议开启。

当线程池无法接受新任务时(通常发生在 shutdown 之后,或者使用了有界队列且队列已满——虽然 DelayedWorkQueue 默认无界,但理论上可以自定义),拒绝策略会被触发。

ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor 的四种标准拒绝策略:

  1. AbortPolicy (默认):直接抛出 RejectedExecutionException 异常,阻止系统正常工作。这是最常用的策略,因为它能明确地告知调用者任务提交失败。
  2. CallerRunsPolicy:不抛弃任务,也不抛出异常,而是将任务回退给调用者线程来执行。这可以降低新任务的提交速率,给线程池喘息的机会。但如果调用者线程(例如主线程或 Tomcat 请求处理线程)繁忙,可能会影响其自身的工作。对于定时任务,这种策略可能不适用,因为它打乱了预期的调度线程。
  3. DiscardPolicy:直接静默丢弃任务,不抛出异常,也没有任何通知。如果任务不重要,允许丢失,可以使用此策略。但存在丢失任务的风险。
  4. DiscardOldestPolicy:丢弃队列中最旧的未处理任务(对于 DelayedWorkQueue 来说,是即将执行的任务),然后尝试重新提交当前任务。这也存在丢失任务的风险。

ScheduledThreadPoolExecutor 的特殊性:

  • 主要在 shutdown 时触发:由于 DelayedWorkQueue 是无界的,正常运行时几乎不会因为队列满而触发拒绝策略。拒绝策略主要在调用 shutdown()shutdownNow() 之后,再尝试提交新任务时发挥作用。
  • delayedExecute 中的检查:在任务入队的代码 delayedExecute 中,有 isShutdown() 检查。如果线程池已关闭,会直接调用 reject() 方法执行配置的拒绝策略。
  • 周期任务的重新入队:周期任务在 reExecutePeriodic 方法中重新入队时,也会检查 canRunInCurrentRunState。如果此时线程池已关闭,任务不会重新入队,相当于被“内部拒绝”了。

实践建议:

  • 对于大多数定时任务场景,默认的 AbortPolicy 是合适的,因为它能让你明确知道任务提交失败(通常是因为线程池已关闭)。
  • 如果你需要在关闭时有特殊的处理逻辑(例如,将未执行的任务持久化),可以实现自定义的 RejectedExecutionHandler 接口。
  • 理解不同策略的行为,根据业务需求选择。避免使用 DiscardPolicyDiscardOldestPolicy,除非你完全确定可以接受任务丢失。

4.3. 线程池的关闭 (shutdown vs shutdownNow)

Section titled “4.3. 线程池的关闭 (shutdown vs shutdownNow)”

优雅地关闭 ScheduledThreadPoolExecutor 对于确保任务的完整性和资源的释放至关重要。有两种关闭方法:

  1. shutdown()

    • 启动平缓的关闭过程。
    • 不再接受新的任务提交(调用 schedule 等方法会触发拒绝策略)。
    • 等待已经提交到队列中的任务(包括延迟任务和周期任务的下一次执行)执行完成
    • 不会中断正在执行的任务。
    • 调用后,isShutdown() 返回 true。当所有任务都完成后,isTerminated() 返回 true
  2. shutdownNow()

    • 启动立即关闭过程。
    • 不再接受新的任务提交。
    • 尝试取消所有等待在队列中的任务。它会遍历队列,调用每个任务的 cancel(true) 方法。cancel(true) 会尝试中断正在执行该任务的线程(如果任务代码响应中断)。
    • 返回等待队列中 未被执行的任务列表 (List<Runnable>)。你可以根据需要处理这些被取消的任务(例如记录日志或尝试重新执行)。
    • 调用后,isShutdown()isTerminated() 通常会很快变为 true(取决于正在执行任务的响应中断情况)。

如何选择?

  • shutdown() (推荐):适用于大多数需要确保已安排任务尽可能完成的场景。这是优雅关闭的标准方式。
  • shutdownNow():适用于需要立即停止所有任务,不关心它们是否完成的场景。例如,应用程序紧急退出。但要注意,它不保证正在执行的任务一定能停止(如果任务不响应中断),且可能导致任务状态不一致。

优雅关闭的最佳实践:

public static void shutdownGracefully(ExecutorService executor, long timeout, TimeUnit unit) {
// 1. 发起 shutdown,不再接受新任务
executor.shutdown();
try {
// 2. 等待一段时间让现有任务执行完成
if (!executor.awaitTermination(timeout, unit)) {
// 3. 如果超时,仍有任务未完成,发起 shutdownNow 尝试强制关闭
System.err.println("线程池未能在 " + timeout + " " + unit.toString().toLowerCase() + " 内关闭,尝试强制关闭...");
List<Runnable> droppedTasks = executor.shutdownNow();
System.err.println("强制关闭,丢弃了 " + droppedTasks.size() + " 个等待中的任务。");
// 4. 再次等待一段时间,让响应中断的任务有机会结束
if (!executor.awaitTermination(timeout / 2, unit)) {
System.err.println("线程池未能最终关闭。");
}
} else {
System.out.println("线程池已成功关闭。");
}
} catch (InterruptedException ie) {
// 5. 如果当前线程在等待过程中被中断,也尝试强制关闭
System.err.println("等待线程池关闭时被中断,尝试强制关闭...");
executor.shutdownNow();
// 重新设置中断状态
Thread.currentThread().interrupt();
}
}
// 使用:
// shutdownGracefully(scheduler, 10, TimeUnit.SECONDS);

这个模式结合了 shutdown() 的优雅和 shutdownNow() 的强制性,并设置了超时等待,是比较健壮的关闭方式。

默认情况下,ScheduledThreadPoolExecutor 创建的线程名字可能是 pool-N-thread-M 这样的格式,不利于问题排查。你可以通过提供自定义的 ThreadFactory 来:

  • 设置有意义的线程名称:例如,包含线程池的业务用途(如 scheduler-report-thread-)。
  • 设置线程为守护线程 (Daemon Thread):如果希望这些线程不阻止 JVM 退出,可以将其设为守护线程。但要注意,如果主线程结束,守护线程会被强制终止,可能导致任务执行到一半。对于重要的定时任务,通常不建议设为守护线程。
  • 设置线程优先级
  • 设置未捕获异常处理器 (UncaughtExceptionHandler):虽然任务内部的异常通常被 ThreadPoolExecutor 捕获,但设置一个总可以捕获其他意外情况。

示例 (使用 Guava 的 ThreadFactoryBuilder):

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
// ...
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("my-scheduler-pool-%d") // %d 会被替换为线程编号
.setDaemon(false) // 设置为非守护线程 (默认)
// .setPriority(Thread.NORM_PRIORITY) // 设置优先级 (默认)
// .setUncaughtExceptionHandler((thread, throwable) -> { // 设置异常处理器
// System.err.println("线程 " + thread.getName() + " 抛出未捕获异常: " + throwable);
// })
.build();
ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(
corePoolSize, // 你设置的核心线程数
namedThreadFactory
// , rejectionHandler // 可以继续指定拒绝策略
);

如果你不想引入 Guava,也可以自己实现 ThreadFactory 接口:

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
class MyThreadFactory implements ThreadFactory {
private final String namePrefix;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final boolean daemon;
public MyThreadFactory(String poolName, boolean daemon) {
this.namePrefix = poolName + "-thread-";
this.daemon = daemon;
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
t.setDaemon(daemon);
// 可以设置 UncaughtExceptionHandler 等
// t.setUncaughtExceptionHandler(...);
return t;
}
}
// 使用:
ThreadFactory myFactory = new MyThreadFactory("report-generator", false);
ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(corePoolSize, myFactory);

使用有意义的线程名对于通过线程 dump 或日志分析问题非常有帮助。

了解线程池的运行状态对于性能调优和问题诊断至关重要。可以通过以下方式监控:

  1. JMX (Java Management Extensions)ThreadPoolExecutor(及其子类 ScheduledThreadPoolExecutor)可以通过 JMX MBean 暴露其内部状态和管理操作。你可以使用 JConsole、VisualVM 或其他 JMX 客户端连接到运行中的 JVM,查看线程池的属性(如 corePoolSize, maximumPoolSize, poolSize, activeCount, largestPoolSize, taskCount, completedTaskCount, queueSize)并执行操作(如修改 corePoolSize)。
  2. ThreadPoolExecutor 提供的 Getter 方法:可以直接在代码中调用 getActiveCount(), getPoolSize(), getQueue().size(), getTaskCount(), getCompletedTaskCount() 等方法获取实时状态。可以将这些指标集成到你的应用监控系统中(如 Prometheus + Grafana)。
  3. Micrometer 等度量库:使用 Micrometer 这样的库可以方便地将线程池指标接入各种监控系统(如 Prometheus, Graphite, Datadog 等)。Micrometer 提供了对 ExecutorService 的绑定,可以自动收集关键指标。

关键监控指标:

  • activeCount: 当前正在执行任务的线程数。如果该值长时间接近 corePoolSize,可能表示线程池处理能力不足。
  • poolSize: 当前线程池中的线程总数。
  • queueSize: DelayedWorkQueue 中等待执行的任务数。持续增长的队列通常意味着 corePoolSize 不足或者任务执行时间过长。
  • taskCount: 已提交到线程池的任务总数(包括已完成、正在执行和等待中的)。
  • completedTaskCount: 已完成执行的任务总数。 taskCount - completedTaskCount - activeCount 约等于 queueSize
  • 任务执行延迟/超时:除了线程池本身的指标,更重要的是监控业务任务是否按时执行。可以在任务执行前后记录时间戳,计算实际执行时间与预期时间的偏差。
特性 java.util.Timer ScheduledThreadPoolExecutor Quartz Spring Task Scheduler (@Scheduled)
线程模型 单线程 多线程 (线程池) 多线程 (线程池) 多线程 (可配置线程池)
异常处理 任务异常导致 Timer 终止 任务异常不影响其他任务,周期任务终止 可配置的任务恢复和重试机制 任务异常默认被记录,不影响其他任务
调度精度 毫秒级,依赖系统时间 纳秒级 (内部),相对时间 毫秒级,支持 Cron 表达式 毫秒级,支持 Cron 表达式、fixedRate/Delay
任务类型 TimerTask (Runnable) Runnable, Callable Job (更复杂,有状态) 方法级注解,Runnable
分布式/集群 不支持 不支持 支持 (通过数据库持久化) 不直接支持 (需借助其他库如 ShedLock)
持久化 不支持 不支持 支持 (任务状态可存数据库) 不支持
依赖 JDK 内置 JDK 内置 需要引入 Quartz 库 需要 Spring Framework (或 Spring Boot)
配置复杂度 非常简单 简单 相对复杂 非常简单 (注解驱动)
适用场景 简单、非关键、单任务场景 (已不推荐) JDK 内置的最常用、灵活的定时任务方案 企业级、需要持久化、集群、复杂调度的场景 Spring 应用中快速实现定时任务

总结:

  • Timer:基本被淘汰,除非在非常老的代码或资源极其有限的环境中。
  • ScheduledThreadPoolExecutor:JDK 自带,功能强大且灵活,适用于绝大多数单机定时任务需求。是 Java 中实现定时任务的基础和标准。
  • Quartz:功能最强大,提供 Cron 表达式、任务持久化、集群支持等高级特性,但配置和使用也更复杂。适用于需要高可用、可恢复、复杂调度的企业级应用。
  • Spring Task Scheduler:基于 ScheduledThreadPoolExecutor (或其他可配置的 TaskScheduler),通过注解 (@Scheduled) 提供了极其方便的使用方式,与 Spring 生态无缝集成。是 Spring 应用中实现定时任务的首选。如果你在使用 Spring,通常会直接用 @Scheduled 而不是手动创建 ScheduledThreadPoolExecutor
  • 定期清理过期 Session/Token:使用 scheduleWithFixedDelay,例如每小时执行一次清理任务。使用 FixedDelay 可以避免因某次清理时间过长影响下次执行。
  • 生成每日统计报表:使用 scheduleAtFixedRate,设置 initialDelay 计算到下一个凌晨(例如凌晨 2 点)的延迟,period 设置为 24 小时。使用 FixedRate 保证每天都在大约同一时间点触发。
  • 服务心跳检测:客户端使用 scheduleAtFixedRate 定期向服务器发送心跳包,服务器也可能需要定时检查客户端连接的活跃状态。
  • 轮询检查外部资源状态:使用 scheduleWithFixedDelay,例如每 5 秒检查一次某个文件是否已生成或某个 API 是否可用。
  • 实现简单的任务重试机制:当某个操作失败时,使用 schedule 安排一个延迟的重试任务。
  • Q: 为什么我的周期任务执行几次后就不执行了?
    • A: 最常见的原因是任务代码抛出了未捕获的异常。如前所述,周期任务抛异常会导致后续执行被取消。请检查任务代码,添加 try-catch 块捕获并记录异常,或者重写 afterExecute 方法来监控异常。
  • Q: scheduleAtFixedRate 的任务执行时间超过了 period 会怎么样?
    • A: 下一次任务会在当前任务结束后立即开始,不会有额外的等待时间。如果任务执行时间持续超过 period,并且线程池大小为 1,任务会连续执行,没有间隔。如果线程池大小大于 1,可能会有多个任务实例并发执行(如果任务允许并发)。
  • Q: 如何取消一个已经提交的定时任务?
    • A: 保存 scheduleXXX 方法返回的 ScheduledFuture 对象。调用其 cancel(boolean mayInterruptIfRunning) 方法。如果参数为 false,则如果任务已开始执行,不会中断它,但后续的周期执行会被取消;如果任务还在队列中等待,则会被移除。如果参数为 true,则会尝试中断正在执行的任务线程(需要任务代码响应 Thread.interrupt()),并取消后续执行。
  • Q: ScheduledThreadPoolExecutor 会导致内存泄漏吗?
    • A: 主要有两种可能:
      • 未关闭线程池:如果创建了 ScheduledThreadPoolExecutor 但从未调用 shutdown()shutdownNow(),线程池的线程(通常是非守护线程)会阻止 JVM 退出,并且持有的资源不会释放。务必在应用退出时关闭线程池。
      • DelayedWorkQueue 无界队列:如果任务提交速度持续快于处理速度,并且任务执行时间很长或 corePoolSize 不足,理论上 DelayedWorkQueue 会无限增长,最终可能导致 OutOfMemoryError。需要合理配置 corePoolSize 并监控队列大小。
  • Q: System.nanoTime() 是否足够精确?它会受 NTP 时间同步影响吗?
    • A: System.nanoTime() 提供纳秒级精度(实际精度取决于操作系统和硬件),主要用于测量时间间隔,它与系统时钟(System.currentTimeMillis())无关,不受 NTP 时间同步或用户手动修改时间的影响。这使得它非常适合用于定时调度。
  • Q: 我应该为每个定时任务创建一个 ScheduledThreadPoolExecutor 吗?
    • A: 不一定。如果任务性质类似,可以共享同一个实例。但如果任务性质差异很大(CPU vs IO,重要 vs 不重要,长耗时 vs 短耗时),或者需要隔离(一个任务的失败不应影响另一个),则可以考虑创建多个实例,并为它们分配合适的资源和配置。

ScheduledThreadPoolExecutor 是 Java 并发包中处理定时和周期性任务的强大武器。它通过结合 ThreadPoolExecutor 的线程管理能力、DelayedWorkQueue 基于时间优先级的任务存储以及 ScheduledFutureTask 对任务时间和周期的封装,提供了一个高效、灵活且健壮的调度框架。

关键要点回顾:

  • 优势:多线程、相对时间、健壮异常处理、灵活调度 (FixedRate vs FixedDelay)。
  • 核心组件DelayedWorkQueue (优先级堆)、ScheduledFutureTask (封装时间、周期、任务逻辑)。
  • 调度方法schedule (单次延迟)、scheduleAtFixedRate (固定速率)、scheduleWithFixedDelay (固定延迟)。
  • 周期任务实现ScheduledFutureTask.run() 在成功执行后,根据 period 正负计算下次时间,并重新入队。
  • 异常处理:任务内异常默认被捕获,不影响线程,但会导致周期任务终止。建议任务内 try-catch 或重写 afterExecute
  • 配置关键:合理设置 corePoolSize 至关重要,需结合任务性质、频率、时长和系统资源考虑。
  • 关闭:务必使用 shutdown()awaitTermination() (或 shutdownNow()) 优雅关闭线程池。
  • 监控:利用 JMX、Getter 方法或 Micrometer 等监控线程池状态。
  • 线程命名:使用自定义 ThreadFactory 提高可维护性。

相比于老旧的 TimerScheduledThreadPoolExecutor 提供了巨大的改进。虽然在 Spring 环境下我们可能更常用 @Scheduled 注解,但理解其底层依赖的 ScheduledThreadPoolExecutor 的工作原理,对于编写高效、可靠的定时任务,以及排查相关问题都大有裨益。
希望本教程能帮助你更深入地理解和应用这个强大的 JUC 工具。

Happy coding!