JUC线程池-ScheduledThreadPoolExecutor
原文:CSDN(历史文章导入,当前状态为草稿)
1. 引言:为什么需要 ScheduledThreadPoolExecutor?
Section titled “1. 引言:为什么需要 ScheduledThreadPoolExecutor?”在 Java 开发中,我们经常会遇到需要延迟执行或周期性执行任务的场景。比如:
- 定时报表生成:每天凌晨生成前一天的业务报表。
- 系统监控:每隔 5 分钟检查一次服务器的健康状态。
- 缓存清理:每小时清理一次过期的缓存数据。
- 任务重试:网络请求失败后,延迟 1 秒后重试。
- 心跳检测:客户端与服务器之间定时发送心跳包维持连接。
对于这些需求,Java 提供了多种解决方案。早期开发者可能会想到 java.util.Timer 和 java.util.TimerTask。然而,Timer 存在一些固有的缺陷,使得它在现代并发应用中不再是首选:
- 单线程执行:
Timer内部仅使用单个后台线程来执行所有任务。如果某个任务执行时间过长,会阻塞后续任务的执行。更糟糕的是,如果某个TimerTask抛出了未捕获的异常,这个唯一的执行线程会终止,导致Timer实例中的所有后续任务都无法再执行。 - 对系统时间敏感:
Timer的调度是基于系统绝对时间的。如果系统时间发生改变(例如,用户手动修改时间),可能会导致任务执行混乱。 - 异常处理不佳:如上所述,未捕获的异常会导致整个
Timer失效,这在健壮性要求高的系统中是不可接受的。
为了克服 Timer 的这些缺点,java.util.concurrent 包(JUC)中引入了 ScheduledThreadPoolExecutor。它是一个功能强大且灵活的线程池实现,专门用于处理定时和周期性任务。
ScheduledThreadPoolExecutor 的核心优势:
- 多线程执行:它是一个真正的线程池,可以使用多个工作线程并发执行任务,避免了单线程瓶颈。
- 相对时间调度:主要基于相对时间进行调度,不易受系统绝对时间变化的影响。
- 健壮的异常处理:单个任务抛出异常不会影响其他任务的执行,也不会导致执行线程终止。线程池内部会捕获异常,并可以选择性地处理它们。
- 更灵活的调度:提供了
scheduleAtFixedRate(固定速率) 和scheduleWithFixedDelay(固定延迟) 两种周期性调度策略,满足不同场景的需求。 - 继承自 ThreadPoolExecutor:拥有
ThreadPoolExecutor的所有优点,如线程管理、拒绝策略、生命周期控制等。
简单来说,ScheduledThreadPoolExecutor 就像一个装备精良、训练有素的“定时任务特种部队”,相比于“单兵作战”的 Timer,它更强大、更可靠、更能适应复杂的并发环境。
本教程将带你深入探索 ScheduledThreadPoolExecutor 的内部世界,从基本用法到核心原理,再到源码分析和最佳实践,助你完全掌握这个强大的并发工具。
2. 核心概念与基础用法
Section titled “2. 核心概念与基础用法”在深入源码之前,我们需要理解 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 的定时调度魔法主要依赖于两个关键组件:
-
DelayedWorkQueue(延迟工作队列):- 这是一个无界(理论上容量是
Integer.MAX_VALUE)的阻塞队列,专门用于存放ScheduledFutureTask。 - 它基于优先级堆(通常是最小堆)实现。队列中的任务并非按照先进先出(FIFO)的顺序排列,而是按照任务的下次执行时间排序,最近需要执行的任务排在队首。
- 只有当任务的延迟时间到了,它才能从队列中被取出 (
take())。如果队首任务的执行时间还未到,尝试获取任务的线程会被阻塞,直到时间到达或被中断。 - 这种设计使得工作线程可以高效地获取即将到期的任务,而无需轮询检查,大大降低了 CPU 消耗。
- 这是一个无界(理论上容量是
-
ScheduledFutureTask(可调度的 Future 任务):- 当你通过
schedule()等方法提交一个Runnable或Callable时,ScheduledThreadPoolExecutor会将其包装成一个ScheduledFutureTask对象。 ScheduledFutureTask继承自FutureTask(提供了获取异步执行结果、取消任务等能力),并实现了RunnableScheduledFuture接口。RunnableScheduledFuture接口又继承了RunnableFuture和Delayed接口。Delayed接口要求实现getDelay(TimeUnit unit)方法,用于返回任务剩余的延迟时间,这是DelayedWorkQueue能够进行时间排序的关键。RunnableScheduledFuture接口还定义了isPeriodic()方法,用于判断任务是否是周期性的。
ScheduledFutureTask内部维护了任务的下次执行时间 (time)、任务的类型(单次、固定速率、固定延迟)以及周期 (period) 等信息。- 它实现了
compareTo()方法,允许DelayedWorkQueue根据执行时间对其进行排序。如果执行时间相同,则根据提交顺序(一个递增的sequenceNumber)排序,保证 FIFO。 - 对于周期性任务,
ScheduledFutureTask的run()方法在执行完任务逻辑后,会重新计算下次执行时间,并将自身重新添加到DelayedWorkQueue中,从而实现周期性调度。
- 当你通过
工作流程简述:
- 用户调用
schedule()、scheduleAtFixedRate()或scheduleWithFixedDelay()提交任务。 ScheduledThreadPoolExecutor将任务包装成ScheduledFutureTask,计算出首次执行时间。ScheduledFutureTask被添加到DelayedWorkQueue中。队列内部根据执行时间(time)使用堆结构进行排序。- 线程池中的工作线程循环调用
DelayedWorkQueue的take()方法尝试获取任务。 - 如果队首任务未到期,
take()方法会阻塞工作线程。 - 当队首任务到期时,
take()方法返回该ScheduledFutureTask。 - 工作线程执行
ScheduledFutureTask的run()方法。 - 如果任务是周期性的,
run()方法内部会计算下次执行时间,并重新将该ScheduledFutureTask添加回DelayedWorkQueue。
这个流程巧妙地结合了线程池、优先级队列和特殊任务封装,实现了高效、可靠的定时调度。
2.3. 三种核心调度方法
Section titled “2.3. 三种核心调度方法”ScheduledExecutorService 接口定义了三种主要的调度方法,ScheduledThreadPoolExecutor 提供了它们的实现:
-
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 秒显示提示信息,或者一个短暂的资源锁定,延迟释放。
- 作用:安排一个任务在指定的
-
scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)- 作用:安排一个任务在指定的
initialDelay之后首次执行,然后以固定的速率 (fixed rate) 重复执行。 - 参数:
command: 要执行的任务。initialDelay: 首次执行的延迟时间。period: 两次执行开始时间之间的间隔。unit:initialDelay和period的时间单位。
- 返回:一个
ScheduledFuture,可以用来取消任务。注意:获取结果通常无意义,因为任务会持续执行。 - 执行时间点计算:
- 首次执行时间点 = 当前时间 +
initialDelay - 第二次执行开始时间点 = 首次执行开始时间点 +
period - 第三次执行开始时间点 = 首次执行开始时间点 + 2 *
period - …
- 第 N 次执行开始时间点 = 首次执行开始时间点 + (N-1) *
period
- 首次执行时间点 = 当前时间 +
- 关键特点:执行时间的计算是基于初始延迟时间点的,与每次任务实际执行花费的时间无关。如果某次任务执行时间超过了
period,那么下次任务会在前一次任务结束后立即开始,试图“追赶”上预定的时间点。这可能导致任务执行的并发(如果线程池大小 > 1)或者连续执行(没有间隔)。 - 场景:适用于需要严格按照固定频率触发的场景,不管任务执行多久。例如:每秒钟刷新一次股票价格,每分钟记录一次系统性能指标。
- 作用:安排一个任务在指定的
-
scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)- 作用:安排一个任务在指定的
initialDelay之后首次执行,然后在每次执行结束之后,再延迟一个固定的时间 (delay) 后重复执行。 - 参数:
command: 要执行的任务。initialDelay: 首次执行的延迟时间。delay: 上一次执行结束到下一次执行开始之间的延迟。unit:initialDelay和delay的时间单位。
- 返回:一个
ScheduledFuture,可以用来取消任务。 - 执行时间点计算:
- 首次执行时间点 = 当前时间 +
initialDelay - 第二次执行开始时间点 = 第一次执行完成时间点 +
delay - 第三次执行开始时间点 = 第二次执行完成时间点 +
delay - …
- 首次执行时间点 = 当前时间 +
- 关键特点:执行时间的计算是基于上一次任务的完成时间点的。它确保了两次任务执行之间至少有
delay的间隔。无论任务执行多长时间,执行间隔是固定的。 - 场景:适用于需要确保两次执行之间有明确间隔的场景,或者任务执行时间不固定,不希望因执行时间过长导致任务堆积。例如:轮询检查某个资源是否可用,每次检查后等待 5 秒再进行下一次检查;或者失败任务的重试,每次重试间隔固定时间。
- 作用:安排一个任务在指定的
选择困难?一张图帮你理解 FixedRate vs FixedDelay:
假设 period 或 delay 都设置为 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 --------------------|假设 period 或 delay 都设置为 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
2.4. 基本用法示例
Section titled “2.4. 基本用法示例”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),观察 FixedRate 和 FixedDelay 在任务执行时间超过 period/delay 时的不同表现。
2.5. 异常处理机制
Section titled “2.5. 异常处理机制”ScheduledThreadPoolExecutor 在异常处理方面比 Timer 健壮得多:
- 工作线程受保护:
ThreadPoolExecutor(ScheduledThreadPoolExecutor的父类)在执行任务的runWorker方法内部有一个大的try-finally块。即使提交的任务本身没有捕获异常,这个try-finally块也会捕获Throwable,确保工作线程不会因为任务抛出的异常而死亡。工作线程在处理完一个(可能异常退出的)任务后,会继续从队列中获取下一个任务执行。 - 周期任务的特殊处理:
- 如果通过
scheduleAtFixedRate或scheduleWithFixedDelay提交的周期性任务在某次执行时抛出了未捕获的异常,那么该任务的后续执行将被取消。它不会再被重新调度。 - 对于单次任务(通过
schedule提交),抛出异常只会导致该次任务执行失败,其Future.get()会抛出ExecutionException。
- 如果通过
- 异常默认被“吞掉”:默认情况下,工作线程捕获到的异常(源自任务内部)不会被打印到控制台,也不会被重新抛出。这可能导致问题被悄无声息地忽略。
- 如何捕获和处理异常:
-
在任务内部
try-catch:最直接的方式是在你的Runnable或Callable的run/call方法内部添加try-catch块,自行处理异常(如记录日志)。这是推荐的做法。 -
重写
afterExecute方法:可以继承ScheduledThreadPoolExecutor并重写protected void afterExecute(Runnable r, Throwable t)方法。这个方法在每个任务执行完毕后被调用。如果任务正常完成,t为null;如果任务抛出异常,t就是那个异常对象。你可以在这里统一处理异常,比如记录日志。注意,需要检查r是否是Future<?>类型,并调用其get()方法来触发潜在的ExecutionException,然后处理t或捕获的ExecutionException。ScheduledThreadPoolExecutor customScheduler = new ScheduledThreadPoolExecutor(1) {@Overrideprotected 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 -
使用
ThreadFactory设置UncaughtExceptionHandler:创建一个自定义的ThreadFactory,为线程池创建的每个线程设置一个UncaughtExceptionHandler。当线程因未捕获的异常而即将终止时(虽然ThreadPoolExecutor的工作线程通常不会因此终止,但设置处理器总是一个好习惯,且可能捕获其他意外情况),这个处理器会被调用。但这对于处理 任务内部 的异常不是主要方式,因为ThreadPoolExecutor通常会捕获它们。
-
强烈建议在任务代码内部处理可预见的异常,并通过 afterExecute 或 try-catch Future.get() 来捕获和记录未预见的异常。
3. 深入源码:解密调度过程
Section titled “3. 深入源码:解密调度过程”理解了基本概念后,让我们深入 ScheduledThreadPoolExecutor 的源码(基于 OpenJDK 11,但核心逻辑在多个版本中类似),看看它是如何精确地实现定时调度的。
3.1. 构造函数与初始化
Section titled “3.1. 构造函数与初始化”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);}关键点分析:
maximumPoolSize固定为Integer.MAX_VALUE:这意味着ScheduledThreadPoolExecutor在理论上可以创建无限多的线程(受系统资源限制)。然而,这通常不是问题,因为任务是由DelayedWorkQueue控制何时执行的。只有当大量任务同时到期时,才可能创建超过corePoolSize的线程。但在实践中,corePoolSize通常是主要的性能瓶颈。keepAliveTime固定为 0:这意味着当线程池中的线程数超过corePoolSize时,多余的空闲线程会立即被终止。但由于maximumPoolSize极大,这个参数在ScheduledThreadPoolExecutor的默认行为下几乎不起作用,除非你自己用其他构造函数或setMaximumPoolSize()强制设置了一个较小的maximumPoolSize。workQueue固定为DelayedWorkQueue:这是实现定时调度的核心队列,后面会详细分析。corePoolSize:这是唯一必须指定的参数,表示线程池中保持活动状态的最小线程数。即使它们是空闲的,也不会被回收(除非设置了allowCoreThreadTimeOut(true),但默认是false)。对于定时任务,保持核心线程活跃通常是必要的,以确保任务能够及时被执行。
思考:为什么 maximumPoolSize 设置为 Integer.MAX_VALUE?
Doug Lea 在设计时可能考虑到,定时任务的执行时机是由任务本身的时间属性决定的,而不是由线程池的繁忙程度。如果大量任务恰好在同一时间点到期,使用有界的 maximumPoolSize 可能会导致任务无法及时执行,甚至被拒绝。将其设置为无界,并将调度的核心压力放在 DelayedWorkQueue 上,可以确保只要系统资源允许,到期的任务总能有线程去执行。同时,keepAliveTime 为 0 确保了这种“超额”创建的线程在空闲后能被快速回收。不过,这也意味着如果配置不当(如 corePoolSize 过小而任务并发度很高),可能会创建过多线程,消耗系统资源。因此,合理设置 corePoolSize 仍然非常重要。
3.2. schedule 方法源码解析
Section titled “3.2. schedule 方法源码解析”我们来看 schedule(Callable<V> callable, long delay, TimeUnit unit) 的实现:
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):
- 参数检查:基本的非空检查。
- 计算触发时间
triggerTime:核心是使用System.nanoTime()获取当前纳秒时间,并加上用户指定的延迟(已转换为纳秒)。nanoTime()用于精确测量时间间隔,是实现准确延迟的关键。使用overflowFree处理可能的 long 溢出。 - 任务包装
decorateTask&ScheduledFutureTask:将用户提交的Callable或Runnable包装成ScheduledFutureTask。这个 task 对象内部存储了计算好的触发时间time。decorateTask是一个钩子方法,允许子类在任务入队前对其进行修改或替换。 - 入队
delayedExecute:- 先检查线程池是否已关闭,若关闭则执行拒绝策略。
- 调用
super.getQueue().add(task)将任务添加到DelayedWorkQueue。队列内部会根据任务的time字段(通过getDelay方法体现)将其放置到优先级堆的正确位置。 - 双重检查关闭状态:添加任务后,再次检查线程池是否关闭。这是一种必要的并发控制,防止在
isShutdown()检查和add()操作之间线程池状态发生改变。如果线程池已关闭,并且该任务(特别是周期性任务)不允许在关闭状态下运行,则尝试从队列中移除该任务并取消它。 - 确保工作线程存在
ensurePrestart:调用ThreadPoolExecutor的ensurePrestart方法。该方法确保至少有一个工作线程启动并运行。如果当前活动线程数小于corePoolSize,或者线程池中没有任何线程,它会创建一个新的工作线程。这个工作线程启动后会立即尝试从DelayedWorkQueue中take()任务。如果队首任务未到期,线程会阻塞等待。
3.3. scheduleAtFixedRate 方法源码解析
Section titled “3.3. scheduleAtFixedRate 方法源码解析”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):
- 参数校验:增加了对
period必须大于 0 的校验。 - 任务包装
ScheduledFutureTask:与schedule类似,但这次构造ScheduledFutureTask时传入了正的period值(转换为纳秒)。ScheduledFutureTask内部会记录这个period。 outerTask引用:sft.outerTask = t;这行代码是为了让内部的ScheduledFutureTask(sft) 能够持有最终被放入队列的(可能被decorateTask装饰过的)任务对象t的引用。这在reExecutePeriodic(处理周期性任务重新入队)和cancel时可能用到。- 入队
delayedExecute:流程与schedule完全相同。
关键在于 ScheduledFutureTask 内部如何处理这个正的 period。我们稍后在分析 ScheduledFutureTask.run() 时会看到。
3.4. scheduleWithFixedDelay 方法源码解析
Section titled “3.4. scheduleWithFixedDelay 方法源码解析”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() 源码解析:周期性调度的核心”ScheduledFutureTask 的 run() 方法是实现周期性调度的关键所在。当工作线程从 DelayedWorkQueue 取出一个到期的 ScheduledFutureTask 并执行其 run() 方法时,会发生以下情况:
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.javavoid 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):
- 区分周期性与非周期性:通过
isPeriodic()(检查period != 0) 判断任务类型。 - 单次任务 (
period == 0):直接调用FutureTask.super.run()。该方法会执行任务,设置结果(或异常),并将Future状态置为完成。任务执行一次后结束。 - 周期性任务 (
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)(基于当前完成时间点)。
- Fixed-Rate (
- 调用
ScheduledThreadPoolExecutor的reExecutePeriodic()方法,该方法内部本质上是再次调用delayedExecute(),将更新了time字段的ScheduledFutureTask重新添加回DelayedWorkQueue。队列会根据新的time值将其放到正确的位置。
- 调用
- 如果
runAndReset返回false(任务抛出异常):任务的Future状态已被设为异常完成,run()方法结束,不会再调用setNextRunTime()和reExecutePeriodic()。因此,周期性任务一旦抛出未捕获的异常,其后续调度就会终止。
- 调用
这就是 ScheduledThreadPoolExecutor 实现周期性调度的核心机制:通过 ScheduledFutureTask 在每次成功执行后重新计算时间并重新入队,实现了任务的循环执行。并通过 period 的正负区分了 FixedRate 和 FixedDelay 的不同计时逻辑。
3.6. DelayedWorkQueue 源码浅析:时间排序的奥秘
Section titled “3.6. DelayedWorkQueue 源码浅析:时间排序的奥秘”DelayedWorkQueue 是 ScheduledThreadPoolExecutor 的心脏。它是一个基于最小堆实现的优先级队列,专门用于存储和管理 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):
- 数据结构:内部使用一个数组
queue来实现最小堆。堆顶queue[0]始终是下次执行时间最早的那个ScheduledFutureTask。 - 排序依据:
siftUp和siftDown操作依赖ScheduledFutureTask实现的compareTo方法进行比较。compareTo首先比较time(下次执行时间),时间早的优先级高;如果时间相同,则比较sequenceNumber(任务提交顺序),序号小的优先级高。 offer(入队):将新任务添加到数组末尾,然后调用siftUp将其“上浮”到堆中正确的位置。如果新任务成为新的堆顶 (queue[0]),则唤醒一个可能在take()中等待的线程。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()唤醒另一个等待的线程,确保队列的处理能够持续进行。
- Leader/Follower 模式优化:引入
- 检查队首
- 并发控制:使用
ReentrantLock和Condition(available) 来保证线程安全和高效的线程协作(等待与唤醒)。
DelayedWorkQueue 通过精巧的堆结构和优化的等待/唤醒机制,实现了让工作线程能够高效、低耗地获取到刚好到期的定时任务。
4. 高级主题与最佳实践
Section titled “4. 高级主题与最佳实践”掌握了核心原理后,我们还需要了解一些高级主题和实践中的注意事项。
4.1. 合理设置线程池大小 (corePoolSize)
Section titled “4.1. 合理设置线程池大小 (corePoolSize)”corePoolSize 是 ScheduledThreadPoolExecutor 最重要的配置参数。设置过小可能导致任务延迟执行,设置过大则会浪费系统资源。如何确定合理的值?
需要考虑的因素:
- 任务的性质:
- CPU 密集型任务:任务需要大量的计算,很少等待 I/O。这种情况下,线程数不宜过多,通常设置为 CPU 核心数 + 1 是一个不错的起点。过多的线程会导致频繁的上下文切换,反而降低性能。
- I/O 密集型任务:任务大部分时间在等待网络、磁盘或其他外部资源响应。例如,调用远程 API、数据库查询、文件读写。这种情况下,线程可以设置得多一些,因为线程在等待时不会占用 CPU。可以设置为 CPU 核心数 * (1 + 平均等待时间 / 平均计算时间),这是一个理论公式,实践中通常设置为 CPU 核心数 * 2 或更多,具体需要根据实际的 I/O 等待情况和系统负载来调整。
- 混合型任务:包含计算和 I/O。需要根据任务中计算和 I/O 的比例来权衡。
- 任务的执行频率和并发度:
- 如果有很多任务需要同时到期并执行,
corePoolSize需要足够大以应对峰值并发。例如,如果预计最多有 10 个任务会同时到期执行,那么corePoolSize至少应设置为 10。 - 如果任务执行非常频繁(例如,毫秒级的周期任务),需要确保有足够的线程来处理,避免任务堆积。
- 如果有很多任务需要同时到期并执行,
- 任务的执行时长:
- 短耗时任务:如果任务执行时间很短,即使并发量大,较小的
corePoolSize可能也能应付。 - 长耗时任务:如果任务执行时间很长,它们会长时间占用线程。如果同时有其他任务到期,就需要更多的线程来避免阻塞。对于非常耗时的周期任务,要特别小心
scheduleAtFixedRate可能导致的执行堆积。
- 短耗时任务:如果任务执行时间很短,即使并发量大,较小的
- 系统资源:服务器的 CPU、内存资源是硬性约束。线程数越多,内存占用(线程栈)和 CPU 上下文切换开销就越大。
- 依赖的外部系统:如果任务依赖于外部服务(如数据库、第三方 API),这些服务的并发处理能力也是一个限制因素。设置过高的线程数可能压垮下游服务。
实践建议:
- 没有银弹公式:最佳线程数通常需要通过性能测试和监控来确定。
- 从保守值开始:可以从一个较小的值(如 CPU 核心数)开始,然后根据应用的性能指标(任务延迟、队列长度、CPU 利用率、线程池活跃线程数等)逐步调整。
- 使用监控工具:利用 JMX、Micrometer、Arthas 等工具监控线程池的各项指标(
activeCount,poolSize,queueSize,taskCount,completedTaskCount),了解其实际运行状况。 - 区分不同类型的任务:如果应用中有多种不同性质的定时任务(CPU密集 vs IO密集,长耗时 vs 短耗时),考虑使用多个
ScheduledThreadPoolExecutor实例,为不同类型的任务配置不同的corePoolSize,避免相互影响。 - 考虑
allowCoreThreadTimeOut(true):如果你希望核心线程在空闲一段时间后也能被回收(适用于任务不频繁的场景,以节省资源),可以调用此方法并设置合适的keepAliveTime。但请注意,这可能导致任务到来时需要重新创建线程,增加一点点延迟。对于需要快速响应的定时任务,通常不建议开启。
4.2. 拒绝策略 (Rejection Policy)
Section titled “4.2. 拒绝策略 (Rejection Policy)”当线程池无法接受新任务时(通常发生在 shutdown 之后,或者使用了有界队列且队列已满——虽然 DelayedWorkQueue 默认无界,但理论上可以自定义),拒绝策略会被触发。
ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor 的四种标准拒绝策略:
AbortPolicy(默认):直接抛出RejectedExecutionException异常,阻止系统正常工作。这是最常用的策略,因为它能明确地告知调用者任务提交失败。CallerRunsPolicy:不抛弃任务,也不抛出异常,而是将任务回退给调用者线程来执行。这可以降低新任务的提交速率,给线程池喘息的机会。但如果调用者线程(例如主线程或 Tomcat 请求处理线程)繁忙,可能会影响其自身的工作。对于定时任务,这种策略可能不适用,因为它打乱了预期的调度线程。DiscardPolicy:直接静默丢弃任务,不抛出异常,也没有任何通知。如果任务不重要,允许丢失,可以使用此策略。但存在丢失任务的风险。DiscardOldestPolicy:丢弃队列中最旧的未处理任务(对于DelayedWorkQueue来说,是即将执行的任务),然后尝试重新提交当前任务。这也存在丢失任务的风险。
ScheduledThreadPoolExecutor 的特殊性:
- 主要在
shutdown时触发:由于DelayedWorkQueue是无界的,正常运行时几乎不会因为队列满而触发拒绝策略。拒绝策略主要在调用shutdown()或shutdownNow()之后,再尝试提交新任务时发挥作用。 delayedExecute中的检查:在任务入队的代码delayedExecute中,有isShutdown()检查。如果线程池已关闭,会直接调用reject()方法执行配置的拒绝策略。- 周期任务的重新入队:周期任务在
reExecutePeriodic方法中重新入队时,也会检查canRunInCurrentRunState。如果此时线程池已关闭,任务不会重新入队,相当于被“内部拒绝”了。
实践建议:
- 对于大多数定时任务场景,默认的
AbortPolicy是合适的,因为它能让你明确知道任务提交失败(通常是因为线程池已关闭)。 - 如果你需要在关闭时有特殊的处理逻辑(例如,将未执行的任务持久化),可以实现自定义的
RejectedExecutionHandler接口。 - 理解不同策略的行为,根据业务需求选择。避免使用
DiscardPolicy和DiscardOldestPolicy,除非你完全确定可以接受任务丢失。
4.3. 线程池的关闭 (shutdown vs shutdownNow)
Section titled “4.3. 线程池的关闭 (shutdown vs shutdownNow)”优雅地关闭 ScheduledThreadPoolExecutor 对于确保任务的完整性和资源的释放至关重要。有两种关闭方法:
-
shutdown():- 启动平缓的关闭过程。
- 不再接受新的任务提交(调用
schedule等方法会触发拒绝策略)。 - 等待已经提交到队列中的任务(包括延迟任务和周期任务的下一次执行)执行完成。
- 不会中断正在执行的任务。
- 调用后,
isShutdown()返回true。当所有任务都完成后,isTerminated()返回true。
-
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() 的强制性,并设置了超时等待,是比较健壮的关闭方式。
4.4. 使用 ThreadFactory 自定义线程
Section titled “4.4. 使用 ThreadFactory 自定义线程”默认情况下,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 或日志分析问题非常有帮助。
4.5. 监控 ScheduledThreadPoolExecutor
Section titled “4.5. 监控 ScheduledThreadPoolExecutor”了解线程池的运行状态对于性能调优和问题诊断至关重要。可以通过以下方式监控:
- JMX (Java Management Extensions):
ThreadPoolExecutor(及其子类ScheduledThreadPoolExecutor)可以通过 JMX MBean 暴露其内部状态和管理操作。你可以使用 JConsole、VisualVM 或其他 JMX 客户端连接到运行中的 JVM,查看线程池的属性(如corePoolSize,maximumPoolSize,poolSize,activeCount,largestPoolSize,taskCount,completedTaskCount,queueSize)并执行操作(如修改corePoolSize)。 ThreadPoolExecutor提供的 Getter 方法:可以直接在代码中调用getActiveCount(),getPoolSize(),getQueue().size(),getTaskCount(),getCompletedTaskCount()等方法获取实时状态。可以将这些指标集成到你的应用监控系统中(如 Prometheus + Grafana)。- Micrometer 等度量库:使用 Micrometer 这样的库可以方便地将线程池指标接入各种监控系统(如 Prometheus, Graphite, Datadog 等)。Micrometer 提供了对
ExecutorService的绑定,可以自动收集关键指标。
关键监控指标:
activeCount: 当前正在执行任务的线程数。如果该值长时间接近corePoolSize,可能表示线程池处理能力不足。poolSize: 当前线程池中的线程总数。queueSize:DelayedWorkQueue中等待执行的任务数。持续增长的队列通常意味着corePoolSize不足或者任务执行时间过长。taskCount: 已提交到线程池的任务总数(包括已完成、正在执行和等待中的)。completedTaskCount: 已完成执行的任务总数。taskCount-completedTaskCount-activeCount约等于queueSize。- 任务执行延迟/超时:除了线程池本身的指标,更重要的是监控业务任务是否按时执行。可以在任务执行前后记录时间戳,计算实际执行时间与预期时间的偏差。
5. 与其他定时任务工具的比较
Section titled “5. 与其他定时任务工具的比较”| 特性 | 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。
6. 实践案例与常见问题
Section titled “6. 实践案例与常见问题”6.1. 实践案例
Section titled “6.1. 实践案例”- 定期清理过期 Session/Token:使用
scheduleWithFixedDelay,例如每小时执行一次清理任务。使用FixedDelay可以避免因某次清理时间过长影响下次执行。 - 生成每日统计报表:使用
scheduleAtFixedRate,设置initialDelay计算到下一个凌晨(例如凌晨 2 点)的延迟,period设置为 24 小时。使用FixedRate保证每天都在大约同一时间点触发。 - 服务心跳检测:客户端使用
scheduleAtFixedRate定期向服务器发送心跳包,服务器也可能需要定时检查客户端连接的活跃状态。 - 轮询检查外部资源状态:使用
scheduleWithFixedDelay,例如每 5 秒检查一次某个文件是否已生成或某个 API 是否可用。 - 实现简单的任务重试机制:当某个操作失败时,使用
schedule安排一个延迟的重试任务。
6.2. 常见问题 (FAQ)
Section titled “6.2. 常见问题 (FAQ)”- Q: 为什么我的周期任务执行几次后就不执行了?
- A: 最常见的原因是任务代码抛出了未捕获的异常。如前所述,周期任务抛异常会导致后续执行被取消。请检查任务代码,添加
try-catch块捕获并记录异常,或者重写afterExecute方法来监控异常。
- A: 最常见的原因是任务代码抛出了未捕获的异常。如前所述,周期任务抛异常会导致后续执行被取消。请检查任务代码,添加
- Q:
scheduleAtFixedRate的任务执行时间超过了period会怎么样?- A: 下一次任务会在当前任务结束后立即开始,不会有额外的等待时间。如果任务执行时间持续超过
period,并且线程池大小为 1,任务会连续执行,没有间隔。如果线程池大小大于 1,可能会有多个任务实例并发执行(如果任务允许并发)。
- A: 下一次任务会在当前任务结束后立即开始,不会有额外的等待时间。如果任务执行时间持续超过
- Q: 如何取消一个已经提交的定时任务?
- A: 保存
scheduleXXX方法返回的ScheduledFuture对象。调用其cancel(boolean mayInterruptIfRunning)方法。如果参数为false,则如果任务已开始执行,不会中断它,但后续的周期执行会被取消;如果任务还在队列中等待,则会被移除。如果参数为true,则会尝试中断正在执行的任务线程(需要任务代码响应Thread.interrupt()),并取消后续执行。
- A: 保存
- Q:
ScheduledThreadPoolExecutor会导致内存泄漏吗?- A: 主要有两种可能:
- 未关闭线程池:如果创建了
ScheduledThreadPoolExecutor但从未调用shutdown()或shutdownNow(),线程池的线程(通常是非守护线程)会阻止 JVM 退出,并且持有的资源不会释放。务必在应用退出时关闭线程池。 DelayedWorkQueue无界队列:如果任务提交速度持续快于处理速度,并且任务执行时间很长或corePoolSize不足,理论上DelayedWorkQueue会无限增长,最终可能导致OutOfMemoryError。需要合理配置corePoolSize并监控队列大小。
- 未关闭线程池:如果创建了
- A: 主要有两种可能:
- Q:
System.nanoTime()是否足够精确?它会受 NTP 时间同步影响吗?- A:
System.nanoTime()提供纳秒级精度(实际精度取决于操作系统和硬件),主要用于测量时间间隔,它与系统时钟(System.currentTimeMillis())无关,不受 NTP 时间同步或用户手动修改时间的影响。这使得它非常适合用于定时调度。
- A:
- Q: 我应该为每个定时任务创建一个
ScheduledThreadPoolExecutor吗?- A: 不一定。如果任务性质类似,可以共享同一个实例。但如果任务性质差异很大(CPU vs IO,重要 vs 不重要,长耗时 vs 短耗时),或者需要隔离(一个任务的失败不应影响另一个),则可以考虑创建多个实例,并为它们分配合适的资源和配置。
ScheduledThreadPoolExecutor 是 Java 并发包中处理定时和周期性任务的强大武器。它通过结合 ThreadPoolExecutor 的线程管理能力、DelayedWorkQueue 基于时间优先级的任务存储以及 ScheduledFutureTask 对任务时间和周期的封装,提供了一个高效、灵活且健壮的调度框架。
关键要点回顾:
- 优势:多线程、相对时间、健壮异常处理、灵活调度 (
FixedRatevsFixedDelay)。 - 核心组件:
DelayedWorkQueue(优先级堆)、ScheduledFutureTask(封装时间、周期、任务逻辑)。 - 调度方法:
schedule(单次延迟)、scheduleAtFixedRate(固定速率)、scheduleWithFixedDelay(固定延迟)。 - 周期任务实现:
ScheduledFutureTask.run()在成功执行后,根据period正负计算下次时间,并重新入队。 - 异常处理:任务内异常默认被捕获,不影响线程,但会导致周期任务终止。建议任务内
try-catch或重写afterExecute。 - 配置关键:合理设置
corePoolSize至关重要,需结合任务性质、频率、时长和系统资源考虑。 - 关闭:务必使用
shutdown()和awaitTermination()(或shutdownNow()) 优雅关闭线程池。 - 监控:利用 JMX、Getter 方法或 Micrometer 等监控线程池状态。
- 线程命名:使用自定义
ThreadFactory提高可维护性。
相比于老旧的 Timer,ScheduledThreadPoolExecutor 提供了巨大的改进。虽然在 Spring 环境下我们可能更常用 @Scheduled 注解,但理解其底层依赖的 ScheduledThreadPoolExecutor 的工作原理,对于编写高效、可靠的定时任务,以及排查相关问题都大有裨益。
希望本教程能帮助你更深入地理解和应用这个强大的 JUC 工具。
Happy coding!