传送门分布式定时任务系列1XXL-job安装分布式定时任务系列2XXL-job使用分布式定时任务系列3任务执行引擎设计分布式定时任务系列4任务执行引擎设计续分布式定时任务系列5XXL-job中blockingQueue的应用分布式定时任务系列6XXL-job触发日志过大引发的CPU告警分布式定时任务系列7XXL-job源码分析之任务触发分布式定时任务系列8XXL-job源码分析之远程调用分布式定时任务系列9XXL-job路由策略分布式定时任务系列10XXL-job源码分析之路由策略番外篇从XXL-job路由策略的“服务容错“说起分布式定时任务系列12XXL-job的任务触发为什么是死循环分布式定时任务系列14XXL-job的注册模型Java并发编程实战1java中的阻塞队列定时任务都是通过死循环触发在前面一共分析过XXL-job的定时任务触发原理、JDK-Timer执行定时任务的原理底层源码都是通过while(true)这种死循环写法来遍历任务的一个是分布式调度框架更为重量级、一个是JDK提供的偏轻量级的调度工具尽管2者定位不同但是都选择了死循环这种方式来实现任务调度说明while(true)一定是可行的、通用的设计方式。但是要问任务调度都是采用这种方式那答案肯定是否定的。一来只分析了XXL-job与Timer的源码导致样本太少二来接下来要分析的另一个任务调度工具也的确不是通过死循环这种方式它就是JDK提供的ScheduledExecutorServiceScheduledExecutorService源码解析ScheduledExecutorService是什么ScheduledExecutorService 是Java并发框架中用于定时任务调度的核心接口基于线程池实现支持延迟任务执行和周期性任务调度相比传统 Timer 类更高效、更灵活。 核心功能提供schedule、scheduleAtFixedRate、scheduleWithFixedDelay等方法可安排任务在指定延迟后执行一次或按固定频率周期执行。支持任务并发执行适用于需要定时轮询、消息推送等场景。 关键特性线程池管理默认使用 ScheduledThreadPoolExecutor 可设置核心线程数如Executors.newScheduledThreadPool(1)支持多任务并发执行 任务调度策略schedule延迟后执行一次任务scheduleAtFixedRate初始延迟后按固定频率重复执行如每5秒执行一次scheduleWithFixedDelay首次执行后等待固定延迟再执行如首次执行后每10秒执行一次 取消任务可通过返回的Future对象取消未执行的任务 适用场景定时轮询数据库或第三方接口如每5分钟检查数据更新定时发送消息或推送通知如每日定时邮件发送周期性任务调度如每2小时重启服务 ScheduledExecutorService的使用ScheduledExecutorService类自jdk1.5才引入作者是大名鼎鼎的Doug LeaJava并发包juc的作者。既然它开始工作的晚自然离退休还早现在一般推荐使用说到Doug Lea自然要推荐他与Joshua Bloch等合著的《Java并发编程实战》见推荐书单。看一下源码中给出的一个例子import static java.util.concurrent.TimeUnit .*; class BeeperControl { // 创建执行器实例初始化1个执行线程 private final ScheduledExecutorService scheduler Executors.newScheduledThreadPool(1); public void beepForAnHour() { final Runnable beeper new Runnable() { public void run() { System.out.println(beep); } }; // 固定频率周期性执行任务每10s执行一次任务发出哗声 final ScheduledFuture? beeperHandle scheduler.scheduleAtFixedRate(beeper, 10, 10, SECONDS); // 延迟 delay 后执行一次任务一次性任务3600s1小时后取消任务 scheduler.schedule(new Runnable() { public void run() { beeperHandle.cancel(true); } }, 60 * 60, SECONDS); } }可以写一个测试跑一下上面的例子然后再用Timer改写上面的例子分别执行会发现效果是一致的import java.util.Date; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; /** * author * date 2025/7/16 */ public class ScheduleTest { private final static ScheduledExecutorService scheduler Executors.newScheduledThreadPool(1); public static void main(String[] args) { ScheduledExecutorServiceBeepForAnHour(); timerBeepForAnHour(); } /** * 使用ScheduledExecutorService实现定时任务 */ public static void ScheduledExecutorServiceBeepForAnHour() { final Runnable beeper new Runnable() { public void run() { System.out.println(new Date() ScheduledExecutorService for beep); } }; final ScheduledFuture? beeperHandle scheduler.scheduleAtFixedRate(beeper, 10, 10, TimeUnit.SECONDS); scheduler.schedule(new Runnable() { public void run() { beeperHandle.cancel(true); } }, 60 * 60, TimeUnit.SECONDS); } /** * 使用Timer实现实时任务 */ public static void timerBeepForAnHour() { // 创建Timer执行器实例 Timer timer new Timer(); // 执行任务发出哗声 TimerTask beeper new TimerTask() { Override public void run() { System.out.println(new Date() Timer for beep); } }; // 固定频率周期性执行任务每10s执行一次任务发出哗声 timer.scheduleAtFixedRate(beeper, 10, 10 * 1000); // 延迟 delay 后执行一次任务一次性任务3600s1小时后取消任务 timer.schedule(new TimerTask() { Override public void run() { beeper.cancel(); } }, 60 * 60 * 1000); } }上面例子要注意的是timer的API里面的时间是毫秒而ScheduledExecutorService是可以通过TimeUnit指定时间单位的。仅通过这个例子就简单的认为功能两者一样那肯定缺乏说服力的。并且如果两个调度器一样也有重复造轮子的嫌疑所以再接着改写一下上面的例子增加一个执行任务执行器里面维持2个任务在任务里面将执行线程信息打印出来但同时将ScheduledExecutorService的执行线程数从1调整为2public class ScheduleTest { // 创建ScheduledExecutorService 执行器 private final static ScheduledExecutorService scheduler Executors.newScheduledThreadPool(2); public static void main(String[] args) { scheduledExecutorServiceBeepForAnHour(); timerBeepForAnHour(); } /** * 使用ScheduledExecutorService实现定时任务 */ public static void scheduledExecutorServiceBeepForAnHour() { // 任务1 final Runnable beeper new Runnable() { public void run() { System.out.println(Thread.currentThread().getName() , new Date() , ScheduledExecutorService for beep1); } }; // 任务2 Runnable beeper2 () - { System.out.println(Thread.currentThread().getName() , new Date() , ScheduledExecutorService for beep2); }; // 添加第1个任务 final ScheduledFuture? beeperHandle scheduler.scheduleAtFixedRate(beeper, 10, 10, TimeUnit.SECONDS); // 添加第2个任务 scheduler.scheduleAtFixedRate(beeper2, 10, 10, TimeUnit.SECONDS); scheduler.schedule(new Runnable() { public void run() { beeperHandle.cancel(true); } }, 60 * 60, TimeUnit.SECONDS); } /** * 使用Timer实现实时任务 */ public static void timerBeepForAnHour() { // 创建Timer执行器实例 Timer timer new Timer(); // 执行任务1发出哗声 TimerTask beeper new TimerTask() { Override public void run() { System.out.println(Thread.currentThread().getName() , new Date() , Timer for beep1); } }; 执行任务2 TimerTask beeper2 new TimerTask() { Override public void run() { System.out.println(Thread.currentThread().getName() , new Date() , Timer for beep2); } }; // 固定频率周期性执行任务每10s执行一次任务发出哗声 timer.scheduleAtFixedRate(beeper, 10, 10 * 1000); timer.scheduleAtFixedRate(beeper2, 10, 10 * 1000); // 延迟 delay 后执行一次任务一次性任务3600s1小时后取消任务 timer.schedule(new TimerTask() { Override public void run() { beeper.cancel(); } }, 60 * 60 * 1000); } }运行程序打印结果Timer-0,Mon Jul 21 11:43:46 CST 2025, Timer for beep1Timer-0,Mon Jul 21 11:43:46 CST 2025, Timer for beep2pool-1-thread-2,Mon Jul 21 11:43:56 CST 2025, ScheduledExecutorService for beep2pool-1-thread-1,Mon Jul 21 11:43:56 CST 2025, ScheduledExecutorService for beep1Timer-0,Mon Jul 21 11:43:56 CST 2025, Timer for beep2Timer-0,Mon Jul 21 11:43:56 CST 2025, Timer for beep1pool-1-thread-1,Mon Jul 21 11:44:06 CST 2025, ScheduledExecutorService for beep1Timer-0,Mon Jul 21 11:44:06 CST 2025, Timer for beep1pool-1-thread-2,Mon Jul 21 11:44:06 CST 2025, ScheduledExecutorService for beep2Timer-0,Mon Jul 21 11:44:06 CST 2025, Timer for beep2pool-1-thread-2,Mon Jul 21 11:44:16 CST 2025, ScheduledExecutorService for beep2pool-1-thread-1,Mon Jul 21 11:44:16 CST 2025, ScheduledExecutorService for beep1上面截取了一部分打印结果会发现对于ScheduledExecutorService来说随着执行线程的增加任务是可能会由不同的线程执行。但是Timer来说它始终只会有一个线程来执行任务。由此我们推断两个类在执行执行上是不同的再改写一下上面的例子将任务阻塞50s大于执行的同期10spublic class ScheduleTest { // 创建执行 private final static ScheduledExecutorService scheduler Executors.newScheduledThreadPool(2); public static void main(String[] args) { scheduledExecutorServiceBeepForAnHour(); timerBeepForAnHour(); } /** * 使用ScheduledExecutorService实现定时任务 */ public static void scheduledExecutorServiceBeepForAnHour() { final Runnable beeper new Runnable() { SneakyThrows public void run() { // 阻塞任务150s TimeUnit.SECONDS.sleep(50); System.out.println(Thread.currentThread().getName() , new Date() , ScheduledExecutorService for beep1); } }; Runnable beeper2 () - { System.out.println(Thread.currentThread().getName() , new Date() , ScheduledExecutorService for beep2); }; // 添加第1个任务 final ScheduledFuture? beeperHandle scheduler.scheduleAtFixedRate(beeper, 10, 10, TimeUnit.SECONDS); // 添加第2个任务 scheduler.scheduleAtFixedRate(beeper2, 10, 10, TimeUnit.SECONDS); scheduler.schedule(new Runnable() { public void run() { beeperHandle.cancel(true); } }, 60 * 60, TimeUnit.SECONDS); } /** * 使用Timer实现实时任务 */ public static void timerBeepForAnHour() { // 创建Timer执行器实例 Timer timer new Timer(); // 执行任务发出哗声 TimerTask beeper new TimerTask() { SneakyThrows Override public void run() { // 阻塞任务150s TimeUnit.SECONDS.sleep(50); System.out.println(Thread.currentThread().getName() , new Date() , Timer for beep1); } }; TimerTask beeper2 new TimerTask() { Override public void run() { System.out.println(Thread.currentThread().getName() , new Date() , Timer for beep2); } }; // 固定频率周期性执行任务每10s执行一次任务发出哗声 timer.scheduleAtFixedRate(beeper, 10, 10 * 1000); timer.scheduleAtFixedRate(beeper2, 10, 10 * 1000); // 延迟 delay 后执行一次任务一次性任务3600s1小时后取消任务 timer.schedule(new TimerTask() { Override public void run() { beeper.cancel(); } }, 60 * 60 * 1000); } }运行程序打印结果pool-1-thread-2,Mon Jul 21 12:20:02 CST 2025, ScheduledExecutorService for beep2pool-1-thread-2,Mon Jul 21 12:20:12 CST 2025, ScheduledExecutorService for beep2pool-1-thread-2,Mon Jul 21 12:20:22 CST 2025, ScheduledExecutorService for beep2pool-1-thread-2,Mon Jul 21 12:20:32 CST 2025, ScheduledExecutorService for beep2pool-1-thread-2,Mon Jul 21 12:20:42 CST 2025, ScheduledExecutorService for beep2Timer-0,Mon Jul 21 12:20:42 CST 2025, Timer for beep1Timer-0,Mon Jul 21 12:20:42 CST 2025, Timer for beep2Timer-0,Mon Jul 21 12:20:42 CST 2025, Timer for beep2pool-1-thread-2,Mon Jul 21 12:20:52 CST 2025, ScheduledExecutorService for beep2pool-1-thread-1,Mon Jul 21 12:20:52 CST 2025, ScheduledExecutorService for beep1从上面的输出观察到的现象对于Timer来说是单线程执行的一旦执行线程被阻塞所有任务都会被阻塞至于阻塞解除才会被重新执行是否补偿取决于不同的API对于ScheduledExecutorService来说是多线程执行的单个线程的阻塞不会造成其它任务的执行理论上执行效率更高ScheduledExecutorService与Timer对比特性TimerScheduledExecutorService引入版本Java 1.3早期Java 5java.util.concurrent包所属包java.util.Timerjava.util.concurrent.ScheduledExecutorService核心功能单线程定时任务调度多线程任务调度支持线程池从构造方法看类结构为了体现ScheduledExecutorService执行器底层用的线程池我们大费周章的写了好几个例子来验证推断其实通过它的构造方法也看出来ScheduledExecutorService scheduler Executors.newScheduledThreadPool(2); // 创建调度线程池 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } // 创建调度线程池这里会super通过调用线程池创建 public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); } // 标准线程池 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } /** * Creates a new {code ThreadPoolExecutor} with the given initial * parameters. * * param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {code allowCoreThreadTimeOut} is set * param maximumPoolSize the maximum number of threads to allow in the * pool * param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * param unit the time unit for the {code keepAliveTime} argument * param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {code Runnable} * tasks submitted by the {code execute} method. * param threadFactory the factory to use when the executor * creates a new thread * param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached * throws IllegalArgumentException if one of the following holds:br * {code corePoolSize 0}br * {code keepAliveTime 0}br * {code maximumPoolSize 0}br * {code maximumPoolSize corePoolSize} * throws NullPointerException if {code workQueue} * or {code threadFactory} or {code handler} is null */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize 0 || maximumPoolSize 0 || maximumPoolSize corePoolSize || keepAliveTime 0) throw new IllegalArgumentException(); if (workQueue null || threadFactory null || handler null) throw new NullPointerException(); this.acc System.getSecurityManager() null ? null : AccessController.getContext(); this.corePoolSize corePoolSize; this.maximumPoolSize maximumPoolSize; this.workQueue workQueue; this.keepAliveTime unit.toNanos(keepAliveTime); this.threadFactory threadFactory; this.handler handler; }由于源码里面涉及到线程池代码比较这里不就完整的贴出来了只展示相关重要的几个类首先创建线程池ScheduledExecutorService scheduler Executors.newScheduledThreadPool(2);Executors介绍Executors的核心作用是封装线程池的创建逻辑通过不同的静态方法提供以下几种线程池类型固定大小的线程池Fixed Thread Pool可缓存的线程池Cached Thread Pool单线程的线程池Single Thread Pool周期性任务调度线程池Scheduled Thread Pool这些线程池分别适用于不同的场景比如任务数量固定、任务数量不确定、需要单线程执行、或需要定时执行任务等。Executors 提供的常用静态方法方法说明适用场景newFixedThreadPool(int nThreads)创建一个固定大小的线程池线程数始终为nThreads任务数量固定、资源有限的场景newCachedThreadPool()创建一个可缓存的线程池线程数可根据任务动态调整任务数量不确定、需要快速响应的场景newSingleThreadExecutor()创建一个单线程的线程池保证任务按顺序执行需要串行执行任务的场景newScheduledThreadPool(int corePoolSize)创建一个支持定时任务调度的线程池需要周期性或延迟执行任务的场景newWorkStealingPool(int parallelism)创建一个基于工作窃取算法的线程池Java 8多核 CPU 下并行任务处理而ScheduledThreadPoolExecutor与ScheduledExecutorService是接口-实现关系而能创建线程池的原因在于ScheduledThreadPoolExecutor与ThreadPoolExecutor的继承关系public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService { // 具体方法略重点明确类关系... }类图直接上类图看看至此得到了ScheduledExecutorService类的完整类图其中通过Executors获取线程池ScheduledThreadPoolExecutorScheduledThreadPoolExecutor有两个内部类DelayedWorkQueue、ScheduledFutureTaskScheduledFutureTask封装了FutureTask可以获取任务执行结果、任务取消、设置任务周期DelayedWorkQueue封装了延迟队列提供了任务管理的方法定时器-任务线程启动通过线程池之后任务的启动就不用定时器自动管理了而是委托给线程池来管理了这里就不展示开有兴趣可以看看之前关于线程池源码的分析《深入解析Java线程池-CSDN博客》同理定时任务执行也不用定时器来触发了定时任务执行谈到定时任务的触发就又回到我们开篇的疑问了是不是所有的调度器都是通过类似while(true)这种死循环策略答案是否定的至少从ScheduledExecutorService代码来看不是这样的它的线程、任务的管理都是通过各种池来实现的这样非常方便。但是线程执行任务完成之后对于周期性的任务还是需要扩展任务、队列来完成比如下面添加任务的代码// 添加周期性任务 public ScheduledFuture? scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { if (command null || unit null) throw new NullPointerException(); if (period 0) throw new IllegalArgumentException(); ScheduledFutureTaskVoid sft new ScheduledFutureTaskVoid(command, null, triggerTime(initialDelay, unit), unit.toNanos(period)); // 封装任务里面存储了任务周期period字段 RunnableScheduledFutureVoid t decorateTask(command, sft); sft.outerTask t; delayedExecute(t); return t; } private class ScheduledFutureTaskV extends FutureTaskV implements RunnableScheduledFutureV { /** Sequence number to break ties FIFO */ private final long sequenceNumber; /** The time the task is enabled to execute in nanoTime units */ private long time; // 任务周期 private final long period; }再继续看延迟执行方法private void delayedExecute(RunnableScheduledFuture? task) { if (isShutdown()) reject(task); else { // 重点是这个添加任务到队列里面 super.getQueue().add(task); if (isShutdown() !canRunInCurrentRunState(task.isPeriodic()) remove(task)) task.cancel(false); else // 添加任务成功调用此方法用于确保线程池中至少有一个工作线程在运行 ensurePrestart(); } }而任务执行完成后如果是同期性任务则会重新设置执行时间来达到周期性执行效果public void run() { // 获取是否周期性任务标志 boolean periodic isPeriodic(); if (!canRunInCurrentRunState(periodic)) cancel(false); // 如果不是周期性任务则直接执行 else if (!periodic) ScheduledFutureTask.super.run(); // 如果是周期性任务则先重置执行状态设置下次执行时间再执行任务 else if (ScheduledFutureTask.super.runAndReset()) { setNextRunTime(); reExecutePeriodic(outerTask); } }加引号的列循环谈到定时任务的触发就又回到我们开篇的疑问了是不是所有的调度器都是通过类似while(true)这种死循环策略答案是否定的在前面回答这个问题时否定加了引号的。严格意义来说线程池底层的代码其实也是一种死循环它通过自旋的方式来获取任务private Runnable getTask() { boolean timedOut false; // Did the last poll() time out? // 没有条件的for循环会一直持有CPU这种方式称为自旋 for (;;) { int c ctl.get(); int rs runStateOf(c); // Check if queue empty only if necessary. if (rs SHUTDOWN (rs STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc workerCountOf(c); // Are workers subject to culling? boolean timed allowCoreThreadTimeOut || wc corePoolSize; if ((wc maximumPoolSize || (timed timedOut)) (wc 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { Runnable r timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r ! null) return r; timedOut true; } catch (InterruptedException retry) { timedOut false; } } }由于里面涉及大量的JAVA并发编程就不展开了可以自行看看源码及相关书籍。推荐书单《Java并发编程实战》力荐五星个人肤浅的认为是看过写的最好的一本Java并发编码方面的书初看惊为天人值的读好几遍不过缺点就是刚看不容易懂也容易忘隔段时间又要重头看起。而且理论偏多代码例子较少有一定的阅读门槛
