Java中定时任务实现方式及源码剖析

概述

在企业级应用开发场景中，定时任务占据着至关重要的地位。比如以下这些场景：

1. 用户4个小时以内没有进行任何操作，就自动清除用户会话。
2. 每天晚上凌晨自动拉取另一个业务系统的某部分数据。
3. 每隔15分钟，自动执行一段逻辑，更新某部分数据。

类似的场景会频繁出现在我们的日常开发中。在Java开发体系中，也有很多实现方案来满足这些场景。但是每个实现方案都有各自的优点和缺点。本文将重点剖析不同实现方案的技术原理以及它们之间的使用场景和差异。

在开始之前，需要先想下定时任务场景的核心技术点是什么？我是这样理解的：

到达未来某一个时间点，执行对应的逻辑。在到达这个时间点之前，需要一直等待。

核心的技术点在于，程序如何知道达到了指定的时间点，然后触发执行对应的逻辑。只要能实现这点，就可以实现定时任务了。本文列举以下六种方案：

1. 循环判定时间
2. Sleep
3. Timer
4. ScheduledExecutorService
5. Spring Scheduling

本文这五种方案的讲解顺序是有考究的，从简单到复杂，从底层到上层。循环判定时间、Sleep旨在摆脱所有组件或者框架带来的复杂度干扰，从最本质上理解定时任务的实现思路。Timer是JDK早先对定时任务的实现，相对来说是比较简单的。ScheduledExecutorService是对Timer的优化、而Spring Scheduling则是基于ScheduledExecutorService实现的。

循环判定时间

我们在线程中，直接使用死循环来不停的判定时间，看是否到了预期的时间点，如果到了就执行逻辑，否则就继续循环。这个方法应该基本不会使用到，但是最能说明定时任务的核心本质。举一个生活化场景的例子：我们请家人帮忙，30分钟后叫自己起床，如果家人不定闹钟的话，他就得不停的去看时间，到了30分钟叫我们起床。

实现

public class LoopScheduler {
    public static void main(String[] avg) {
        long nowTime = System.currentTimeMillis();
        long nextTime = nowTime + 15000;
        while (true) {
            if (System.currentTimeMillis() >= nextTime) {
                nowTime = System.currentTimeMillis();
                nextTime = nowTime + 15000;
                System.out.println(nowTime + ":触发一次");
                service();
            }
        }
    }
public static void service() {
    System.out.println("自定义逻辑执行");
}
}

以上代码就可以实现每隔15s执行service方法一次。以下是执行情况：

可以看到确实非常严格的15s执行一次。实现了定时任务的效果。

分析

这种实现方式之所以基本不会实际使用，是因为这个while循环的空转会占用非常多宝贵的cpu资源。但是可以借此看到定时任务的实现框架。

Sleep

借助于Thread.sleep()，我们可以实现让线程等待指定时间后再执行。Thread.sleep()方法是一个JNI方法，其底层是与操作系统内核进行交互。调用该方法，可以将线程进入睡眠状态，在这种状态中，该线程不会获取到cpu资源。直到指定的睡眠时间结束，操作系统会根据调度策略将线程唤醒。

实现

public class SleepScheduler {

public static void main(String[] avg) {

Thread thread = new Thread(new Runnable() {

@Override

public void run() {

while (true) {

try {

Thread.sleep(15000);

System.out.println(System.currentTimeMillis() + “:触发一次”);

service();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}        public void service() {
            System.out.println("自定义逻辑执行");
        }
    });
    thread.start();
}
}

以上代码，先定义了一个线程，然后启动。线程的执行逻辑是，不断循环，每次循环里面先sleep 15s。然后再执行指定的逻辑。基本上可以实现跟上面一样的效果，每15s执行一次service逻辑。

我们观察下执行的情况就会发现，每次执行的间隔并不是严格的15s。一般都会比15s要多一点。这是因为sleep的机制导致的。sleep结束之后，线程并不会立马获得执行，线程只是会被重新放入调度队列参与下一次调度。

分析

使用Thread.sleep()跟我们自己用循环去判断时间相比，最大的优势在于我们节省了CPU资源。利用操作系统的线程调度能力去实现对时间的控制和判断。

Timer

Timer是JDK自带的调度工具类，针对定时任务的场景场景已经做了抽象和封装。Timer的核心入口是Timer类的schedule方法。用于提交一个任务，并且可以指定延迟时间和重复间隔。

/**

* @param task   task to be scheduled.

* @param delay  delay in milliseconds before task is to be executed.

* @param period time in milliseconds between successive task executions.

*/

public void schedule(TimerTask task, long delay, long period) {

if (delay < 0)

throw new IllegalArgumentException(“Negative delay.”);

if (period <= 0)

throw new IllegalArgumentException(“Non-positive period.”);

sched(task, System.currentTimeMillis()+delay, -period);

}

如何使用

public class TimerScheduler {public static void main(String[] args) {
    // 创建SimpleDateFormat对象，定义格式化样式
    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
// 创建Timer对象
Timer timer = new Timer();

// 单次执行任务示例
TimerTask singleTask = new TimerTask() {
    @Override
    public void run() {
        // 将当前时间的毫秒数转换为格式化后的时间字符串
        String formattedTime = sdf.format(System.currentTimeMillis());
        System.out.println(formattedTime + "单次执行任务：定时任务执行了");
    }
};
// 延迟3000毫秒（3秒）后执行单次任务
timer.scheduleAtFixedRate(singleTask, 3000, 15000);
String formattedTime = sdf.format(System.currentTimeMillis());
System.out.println(formattedTime + "单次执行任务：定时任务已启动");
}
}

以上代码首先创建了一个TimerTask的实例，然后重写run方法封装我们的业务逻辑，然后调用Timer的scheduleAtFixedRate方法将任务进行提交。提交之后3s之后开始执行一次，然后以15s执行一次的固定频率不断执行下去。

Timer一共提供了6个api来提交任务，可以分为三大类：

• 下图1和2：需要传入period的schedule()，提交的是周期任务，period是周期。不过这里的周期是上一次任务成功结束和下一次任务开始执行的间隔。可以认为是按照固定的延迟去重复。
• 下图3和4：不需要传入period的schedule()，提交的是单次任务。
• 下图5和6：scheduleAtFixedRate()，提交的是周期任务，period是执行频率，每次任务的计划执行时间在提交的那一刻就可以确定，跟上一次任务什么时候结束没有关系。

原理解析

Timer已经是JDK针对定时任务场景做了抽象和封装后的成果，其核心实现类除了Timer、TimerTask。还有Timer的内部类TaskQueue、TimerThread。

TimerTask

public abstract class TimerTask implements Runnable {

/**

* This object is used to control access to the TimerTask internals.

*/

final Object lock = new Object();/**
 * The state of this task, chosen from the constants below.
 */
int state = VIRGIN;
/**This task has not yet been scheduled.

*/

static final int VIRGIN = 0;/**This task is scheduled for execution.  If it is a non-repeating task,it has not yet been executed.

*/

static final int SCHEDULED   = 1;/**This non-repeating task has already executed (or is currentlyexecuting) and has not been cancelled.

*/

static final int EXECUTED    = 2;/**This task has been cancelled (with a call to TimerTask.cancel).

*/

static final int CANCELLED   = 3;/**Next execution time for this task in the format returned bySystem.currentTimeMillis, assuming this task is scheduled for execution.For repeating tasks, this field is updated prior to each task execution.

*/

long nextExecutionTime;/**Period in milliseconds for repeating tasks.  A positive value indicatesfixed-rate execution.  A negative value indicates fixed-delay execution.A value of 0 indicates a non-repeating task.

*/

long period = 0;


}

顾名思义，TimerTask是对任务的抽象，已经定义好了任务状态、是否可重复（period）、下次执行时间等属性。除此之外，TimerTask实现了Runnable，我们可以通过重写run方法来封装我们的业务逻辑。

Timer

Timer是对调度器的抽象。其暴露了schedule核心api用来让我们进行任务提交。它有两个比较重要的内部类，是理解Timer实现原理的关键。

public class Timer {

/**

* The timer task queue.  This data structure is shared with the timer

* thread.  The timer produces tasks, via its various schedule calls,

* and the timer thread consumes, executing timer tasks as appropriate,

* and removing them from the queue when they’re obsolete.

*/

private final TaskQueue queue = new TaskQueue();/**
 * The timer thread.
 */
private final TimerThread thread = new TimerThread(queue);
public Timer(String name) {

thread.setName(name);

thread.start();

}
}

在Timer的构造方法中，会直接将TimerThread启动。

TaskQueue

TaskQueue内部持有一个TimerTask数组。用来存储通过schedule方法提交的TimerTask实例。实际上TimerTask是一个小顶堆，每一次添加元素的时候，都会根据任务的下次执行时间(nextExecutionTime)来维护小顶堆，将下次执行时间最近的任务放在堆顶。

class TaskQueue {

/**

* Priority queue represented as a balanced binary heap: the two children

* of queue[n] are queue[2n] and queue[2n+1].  The priority queue is

* ordered on the nextExecutionTime field: The TimerTask with the lowest

* nextExecutionTime is in queue[1] (assuming the queue is nonempty).  For

* each node n in the heap, and each descendant of n, d,

* n.nextExecutionTime <= d.nextExecutionTime.

*/

private TimerTask[] queue = new TimerTask[128];/**
 * The number of tasks in the priority queue.  (The tasks are stored in
 * queue[1] up to queue[size]).
 */
private int size = 0;
}

TimerThread

TimerThread继承了Thread。会在Timer类初始化的时候直接启动。所以核心要关注的是TimerThread的run方法，run方法的主体是调用mainLoop()方法。

class TimerThread extends Thread {/**
 * Our Timer's queue.  We store this reference in preference to
 * a reference to the Timer so the reference graph remains acyclic.
 * Otherwise, the Timer would never be garbage-collected and this
 * thread would never go away.
 */
private TaskQueue queue;
TimerThread(TaskQueue queue) {

this.queue = queue;

}public void run() {

try {

mainLoop();

} finally {

// Someone killed this Thread, behave as if Timer cancelled

synchronized(queue) {

newTasksMayBeScheduled = false;

queue.clear();  // Eliminate obsolete references

}

}

}

private void mainLoop() {

while (true) {

try {

TimerTask task;

boolean taskFired;

synchronized(queue) {

// Wait for queue to become non-empty

while (queue.isEmpty() && newTasksMayBeScheduled)

queue.wait();

if (queue.isEmpty())

break; // Queue is empty and will forever remain; die            // Queue nonempty; look at first event and do the right thing
            long currentTime, executionTime;
            task = queue.getMin();
            synchronized(task.lock) {
                if (task.state == TimerTask.CANCELLED) {
                    queue.removeMin();
                    continue;  // No action required, poll queue again
                }
                currentTime = System.currentTimeMillis();
                executionTime = task.nextExecutionTime;
                if (taskFired = (executionTime<=currentTime)) {
                    if (task.period == 0) { // Non-repeating, remove
                        queue.removeMin();
                        task.state = TimerTask.EXECUTED;
                    } else { // Repeating task, reschedule
                        queue.rescheduleMin(
                          task.period<0 ? currentTime   - task.period
                                        : executionTime + task.period);
                    }
                }
            }
            if (!taskFired) // Task hasn't yet fired; wait
                queue.wait(executionTime - currentTime);
        }
        if (taskFired)  // Task fired; run it, holding no locks
            task.run();
    } catch(InterruptedException e) {
    }
}
}
}

mainLoop主体逻辑如下：

1. 主体是一个while循环，进入循环后，会先看任务队列是否有任务，如果没有任务则执行quere.wait()。等待其他线程(一般是主线程在添加任务后)执行queue.notify()方法后，TimerThread会再次醒来。
2. 如果队列不为空，成功获取到堆顶元素后，会判断任务是否已经被取消，如果取消的话直接移除，然后进入下一次循环。
3. 如果任务没有被取消，则看任务的执行时间是否已经到了，也就是该任务是否应该被触发了。如果已经触发了，那么应该执行该任务，同时如果该任务是周期任务，还应该计算下次任务的执行时间，然后触发堆的下滤操作（重新找一个下次执行时间最近的任务到堆顶）。
4. 如果任务还没有到执行时间，那么就计算还剩多少时间应该执行，然后等待这个时间。重新进入下一次循环。

细心的同学，应该可以注意到，在处理重复任务的时候，会判断period是负数还是正数。

负数的话，任务的下次执行时间是：currentTime - task.period（当前任务结束的时间 - 周期）

正数的话，任务的下次执行时间是：executionTime + task.period（当前任务的执行时间 + 周期）

但是我们传入的period不是都是正数吗？可以看下面代码，schedule在调用sched方法的时候对我们的period取负了。可以这样理解，period同时也是一个任务类型的标识，period为0表示的是单次任务。period为负数就表示的是按照固定延迟去重复的任务，period为正数表示的是按照固定频率去重复的任务。

public void schedule(TimerTask task, long delay, long period) {

if (delay < 0)

throw new IllegalArgumentException(“Negative delay.”);

if (period <= 0)

throw new IllegalArgumentException(“Non-positive period.”);

sched(task, System.currentTimeMillis()+delay, -period);

}public void scheduleAtFixedRate(TimerTask task, long delay, long period) {

if (delay < 0)

throw new IllegalArgumentException(“Negative delay.”);

if (period <= 0)

throw new IllegalArgumentException(“Non-positive period.”);

sched(task, System.currentTimeMillis()+delay, period);

}

综合分析

1. Timer是基于Object.wait()方法来实现时间等待的。在本质上跟我们用Thread.sleep()实现没有太大的区别，底层都是通过操作系统内核让线程进入睡眠状态。
2. Timer是单线程模式，通过上面的分析可以看到，Timer类持有一个TimerThread实例，它就是一个线程实例。这种模式的问题在于，当我们在同一个Timer中提交了很多调度任务之后，并且有的任务时间过长的话，就可能会导致其他任务的调度延后。

注意事项

1. 通过上面的源码分析可以看到，TimerThread执行task.run()的时候，没有进行异常处理，所以当使用Timer的时候，业务逻辑一定要进行异常处理，否则如果一旦抛出异常将导致TimerThread线程挂掉。所有调度任务就失效了。

ScheduledExecutorService

ScheduledExecutorService也是JDK自带的定时任务工具，是在JDK1.5引入的。可以理解为是对Timer的升级，通过上文我们知道：Timer是基于单线程模型的，必然跟不上Java领域开发的发展步伐。而ScheduledExecutorService是基于线程池来实现的。这也是对Timer最主要的优化，当我们理解了Timer之后，再看ScheduledExecutorService就会简单很多。

如何使用

public class ScheduldExecutorServiceScheduler {

public static void main(String[] args) {

// 创建SimpleDateFormat对象，定义格式化样式

SimpleDateFormat sdf = new SimpleDateFormat(“HH:mm:ss.SSS”);

// 创建ScheduledExecutorService对象

ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

Runnable runnable = new Runnable() {

@Override

public void run() {

// 将当前时间的毫秒数转换为格式化后的时间字符串

String formattedTime = sdf.format(System.currentTimeMillis());

System.out.println(formattedTime + “单次执行任务：定时任务执行了”);

}

};

// 延迟3000毫秒（3秒）后执行单次任务，每15s开始执行一次任务

executor.scheduleAtFixedRate(runnable, 3000, 15000, java.util.concurrent.TimeUnit.MILLISECONDS);

// 延迟3000毫秒（3秒）后执行单次任务，每次任务结束后15s执行下一次任务
//executor.scheduleWithFixedDelay(runnable, 3000, 15000, java.util.concurrent.TimeUnit.MILLISECONDS);

// 延迟3000毫秒（3秒）后执行单次任务
//executor.schedule(runnable, 3000, java.util.concurrent.TimeUnit.MILLISECONDS);

String formattedTime = sdf.format(System.currentTimeMillis());

System.out.println(formattedTime + “单次执行任务：定时任务已启动”);

}}

上述代码通过Executors创建了一个单线程的调度执行器。然后通过调用执行器的scheduleAtFixedRate()方法提交封装好的Runnable任务，跟前文的例子一致，也是延迟3s后执行，然后按照15s的固定频率去重复执行。

下图是执行效果。

除此之外，ScheduledExecutorService也暴露了scheduleWithFixedDelay（固定延迟）、schedule方法（单次执行）。与上文我们分析的Timer的API是一致的。只不过scheduleWithFixedDelay的api更加语义化了。

如果需要多线程的执行器，那么使用Executors.newScheduledThreadPool()方法创建指定线程数量的线程池执行器。

public static void main(String[] args) {

// 创建SimpleDateFormat对象，定义格式化样式

SimpleDateFormat sdf = new SimpleDateFormat(“HH:mm:ss.SSS”);

// 创建ScheduledExecutorService对象

//ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

Runnable runnable = new Runnable() {

@Override

public void run() {

// 将当前时间的毫秒数转换为格式化后的时间字符串

String formattedTime = sdf.format(System.currentTimeMillis());

System.out.println(formattedTime + “单次执行任务：定时任务执行了”);

}

};

// 延迟3000毫秒（3秒）后执行单次任务，每15s开始执行一次任务

executor.scheduleAtFixedRate(runnable, 3000, 15000, java.util.concurrent.TimeUnit.MILLISECONDS);

// 延迟3000毫秒（3秒）后执行单次任务，每次任务结束后15s执行下一次任务
//executor.scheduleWithFixedDelay(runnable, 3000, 15000, java.util.concurrent.TimeUnit.MILLISECONDS);

// 延迟3000毫秒（3秒）后执行单次任务
//executor.schedule(runnable, 3000, java.util.concurrent.TimeUnit.MILLISECONDS);

String formattedTime = sdf.format(System.currentTimeMillis());

System.out.println(formattedTime + “单次执行任务：定时任务已启动”);

}

原理解析

在解读Timer的源码的时候，我们知道有这几个核心角色：

TimerTask：任务，对具体任务逻辑的抽象和封装。

Timer：调度器，负责所有任务的提交和注册。

TaskQueue：任务队列，Timer对象持有，通过Timer提交的任务都存在这个队列中

TimerThread：单线程执行器，Timer对象持有，TimerThread会不断地从TaskQueue中获取任务来执行。

在ScheduledExecutorService中，这几个角色同样存在，只不过换了名字。并且在技术层面做了优化。

ScheduledFutureTask：对任务的封装。

ScheduledExecutorService：调度器，负责所有任务的提交和注册。

BlockingQueue：阻塞队列，用来存放任务，与普通队列不同的是，当尝试从阻塞队列中获取元素时，如果队列为空，那么线程会阻塞。

ThreadPoolExecutor：线程池执行器。每个线程都会不断地从任务队列中尝试获取任务并执行。

ScheduledFutureTask

根据ScheduledFutureTask的UML图可以看到，ScheduledFutureTask主要继承了FutureTask类。FutureTask是java异步编程的一个核心类，这里暂不展开。先简单理解为他对异步任务的执行生命周期、返回结果做了封装。我们可以通过FutureTask的api去判断一个任务是否执行完成、取消任务、获取执行结果等操作。

下面是ScheduledFutureTask的核心部分代码，为了理清主要脉络，对代码做了删减。

可以看到的是，ScheduledFutureTask跟前文的Timer一样，有任务的周期、下次执行时间等属性。不同的是ScheduledFutureTask有自己的run方法。 在Timer中，TimerTask的run方法是我们自己重写的，就是业务逻辑代码。在ScheduledExecutorService中，我们把Runnable提交给ScheduledExecutorService之后，则是先调用ScheduledFutureTask的构造方法，在ScheduledFutureTask的构造方法中，又调用父类（FutureTask）的构造方法，并传入Runnable实例。

private class ScheduledFutureTask<V>

extends FutureTask<V> implements RunnableScheduledFuture<V> {    /** The time the task is enabled to execute in nanoTime units */
    private long time;
/**
 * Period in nanoseconds for repeating tasks.  A positive
 * value indicates fixed-rate execution.  A negative value
 * indicates fixed-delay execution.  A value of 0 indicates a
 * non-repeating task.
 */
private final long period;


/**
 * Creates a one-shot action with given nanoTime-based trigger time.
 */
ScheduledFutureTask(Runnable r, V result, long ns) {
    super(r, result);
    this.time = ns;
    this.period = 0;
    this.sequenceNumber = sequencer.getAndIncrement();
}

/**
 * Overrides FutureTask version so as to reset/requeue if periodic.
 */
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);
    }
}
}</code></pre><p style="">这里run方法的逻辑是：</p><p style=""><img src="https://halo-www-twofish-site.oss-cn-beijing.aliyuncs.com/attachments/uploads/2024/11/ScheduledFutureTask的run方法流程.png" width="100%" height="100%" style="display: inline-block"></p><ol><li><p style="">判断任务是否能执行，因为存在已经将调度器shutdown了的情况。</p></li><li><p style="">如果是非周期任务，那么执行即可</p></li><li><p style="">如果是周期任务，除了要执行之外，还需要重新安排下一次调度。</p></li></ol><h4 style="" id="scheduledexecutorservice-1">ScheduledExecutorService</h4><p style="">前面我们已经提过，我们通过ScheduledExecutorService暴露的api提交任务，然后由ScheduledExecutorService持有的阻塞队列实例去存储提交上来的任务。另外就是通过线程池去执行任务，而不是单线程。本文就暂时不展开阻塞队列了，因为跟本文的主题已经有点远了。我们直接介绍线程池这部分。</p><p style=""><img src="https://halo-www-twofish-site.oss-cn-beijing.aliyuncs.com/attachments/uploads/2024/11/20241119-M8XCagQBvO.png" width="100%" height="100%" style="display: inline-block"></p><p style="">ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，所以其本身也是线程池实例。线程池中的线程是随我们任务的提交，不断创建的。以scheduleAtFixedRate为例：提交任务后，先是调用delayedExecute方法。</p><pre><code class="language-java">// ScheduledThreadPoolExecutor.java
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();

ScheduledFutureTask<Void> sft =

new ScheduledFutureTask<Void>(command,

null,

triggerTime(initialDelay, unit),

unit.toNanos(period));

RunnableScheduledFuture<Void> t = decorateTask(command, sft);

sft.outerTask = t;

delayedExecute(t);

return t;

}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();

}

}

最终会调用到父类的ensurePrestart方法，这个方法会获取当前Worker的数量，如果Worker的数量小于设定的大小，那么就增加Worker。

// ThreadPoolExecutor.java

void ensurePrestart() {

int wc = workerCountOf(ctl.get());

if (wc < corePoolSize)

addWorker(null, true);

else if (wc == 0)

addWorker(null, false);

}

Worker

Worker实现了Runnerable接口，并且持有一个Thread实例。在添加Worker的时候，会直接将Worker持有的线程启动，并且执行Worker的run方法。

private final class Worker

extends AbstractQueuedSynchronizer

implements Runnable

{

/**

* This class will never be serialized, but we provide a

* serialVersionUID to suppress a javac warning.

*/

private static final long serialVersionUID = 6138294804551838833L;    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;
/**
 * Creates with given first task and thread from ThreadFactory.
 * @param firstTask the first task (null if none)
 */
Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

/** Delegates main run loop to outer runWorker  */
public void run() {
    runWorker(this);
}</code></pre><p style="">可以看到，run方法只有一行代码，就是将自己的引用作为入参调用runWorker方法。runWorker是ThreadPoolExecutor类的方法。</p><pre><code class="language-java">final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
    while (task != null || (task = getTask()) != null) {
        w.lock();
        // If pool is stopping, ensure thread is interrupted;
        // if not, ensure thread is not interrupted.  This
        // requires a recheck in second case to deal with
        // shutdownNow race while clearing interrupt
        if ((runStateAtLeast(ctl.get(), STOP) ||
             (Thread.interrupted() &&
              runStateAtLeast(ctl.get(), STOP))) &&
            !wt.isInterrupted())
            wt.interrupt();
        try {
            beforeExecute(wt, task);
            Throwable thrown = null;
            try {
                task.run();
            } catch (RuntimeException x) {
                thrown = x; throw x;
            } catch (Error x) {
                thrown = x; throw x;
            } catch (Throwable x) {
                thrown = x; throw new Error(x);
            } finally {
                afterExecute(task, thrown);
            }
        } finally {
            task = null;
            w.completedTasks++;
            w.unlock();
        }
    }
    completedAbruptly = false;
} finally {
    processWorkerExit(w, completedAbruptly);
}
}</code></pre><ol><li><p style="">主体不出意料的是个while循环，如果获取不到任务，就不进入循环体，而是直接退出，然后走processWorkerExit，移除worker。</p></li><li><p style="">如果获取到任务的话，进入循环体，先判断线程池是否已经要被关闭。如果状态正常，就执行任务，运行task.run()方法，也就是我们前面说的ScheduledFutureTask的run方法。</p></li></ol><h3 style="" id="%E7%BB%BC%E5%90%88%E5%88%86%E6%9E%90-1">综合分析</h3><ol><li><p style="">ScheduledThreadPoolExecutor在实现的复杂度上比Timer大了不少，用到了阻塞队列、线程池。从功能和性能角度都有很大的优化空间。</p></li><li><p style="">在异常处理层面，我们可以看到也很完善，不用再担心因为忘记处理的异常导致整个线程挂掉而影响后续的任务执行。</p></li></ol><h2 style="" id="spring-scheduling">Spring Scheduling</h2><p style="">Spring Scheduling是Spring提供的定时任务工具，与ScheduledThreadPoolExecutor相比，易用性上有了质的飞跃。</p><ol><li><p style="">只需要写好业务方法，然后在方法上添加上@Scheduled注解，配置好定时策略，所有工作就完成了。</p></li><li><p style="">提供更复杂的调度策略，比如cron表达式。</p></li></ol><h3 style="" id="%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8-2">如何使用</h3><ol><li><p style="">在SpringBoot的主程序上添加@EnableScheduling注解。</p></li></ol><pre><code>@SpringBootApplication
@EnableScheduling

public class SpringbootsrcApplication {public static void main(String[] args) {
    SpringApplication.run(SpringbootsrcApplication.class, args);
}
}

1. 在业务方法上添加@Scheduled注解。

@Component

public class SpringScheduler {// 创建SimpleDateFormat对象，定义格式化样式
private SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
@Scheduled(cron = "0/15 * * * * ? ")

public void service() throws InterruptedException {

// 将当前时间的毫秒数转换为格式化后的时间字符串

String formattedTime = sdf.format(System.currentTimeMillis());

System.out.println(formattedTime + "单次执行任务：定时任务执行了");

}
}

值得注意的是，类上面的@Componet注解不能少，必须要将Bean注册给Spring，Spring的调度功能才能生效。

以上就完成了一个使用cron定义的定时任务的开发，从第0s开始，每15s执行一次。

@Scheduled还支持其他的参数，其中就包括我们非常熟悉的initialDelay、fixedDelay、FixedRate。

public @interface Scheduled {String cron() default "";
String zone() default "";long fixedDelay() default -1;String fixedDelayString() default "";long fixedRate() default -1;String fixedRateString() default "";long initialDelay() default -1;String initialDelayString() default "";TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
}

原理解析

Spring根据我们@Scheduled上的配置，自动为我们管理好了调度任务。可以想到的是，它必然要收集到这些信息才行。这就引出了Spring Scheduling的第一个核心类。

ScheduledAnnotationBeanPostProcessor

ScheduledAnnotationBeanPostProcessor实现了非常多的接口，但是我们可以只用看红框圈出的这两个。

BeanPostProcessor：有两个接口方法postProcessBeforeInitialization、postProcessAfterInitialization。分别是Bean初始化的前置处理器和后置处理器。

ApplicationListener：用于监听Spring生命周期中的各种事件。

ScheduledAnnotationBeanPostProcessor通过实现postProcessAfterInitialization方法，在每个bean初始化后，对bean进行扫描，看是否存在被@Scheduled注解的方法。对有@Scheduled注解的方法进行遍历，分别执行processScheduled方法。

@Override

public Object postProcessAfterInitialization(Object bean, String beanName) {

if (bean instanceof AopInfrastructureBean || bean instanceof TaskScheduler ||

bean instanceof ScheduledExecutorService) {

// Ignore AOP infrastructure such as scoped proxies.

return bean;

}Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
if (!this.nonAnnotatedClasses.contains(targetClass) &&
      AnnotationUtils.isCandidateClass(targetClass, Arrays.asList(Scheduled.class, Schedules.class))) {
   Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
         (MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {
            Set<Scheduled> scheduledAnnotations = AnnotatedElementUtils.getMergedRepeatableAnnotations(
                  method, Scheduled.class, Schedules.class);
            return (!scheduledAnnotations.isEmpty() ? scheduledAnnotations : null);
         });
   if (annotatedMethods.isEmpty()) {
      this.nonAnnotatedClasses.add(targetClass);
      if (logger.isTraceEnabled()) {
         logger.trace("No @Scheduled annotations found on bean class: " + targetClass);
      }
   }
   else {
      // Non-empty set of methods
      annotatedMethods.forEach((method, scheduledAnnotations) ->
            scheduledAnnotations.forEach(scheduled -> processScheduled(scheduled, method, bean)));
      if (logger.isTraceEnabled()) {
         logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +
               "': " + annotatedMethods);
      }
   }
}
return bean;
}

processScheduled方法比较长，但是逻辑其实很简单：

1. 首先基于获取到的bean、method，利用反射技术，封装为一个Runnable实例。
2. 根据@Schedule的不同参数配置，识别出不同类型的任务：cron表达式任务、fixedDelay、fixedRate任务。然后通过registry的不同api，提交这些任务。

protected void processScheduled(Scheduled scheduled, Method method, Object bean) {

try {

Runnable runnable = createRunnable(bean, method);

boolean processedSchedule = false;

String errorMessage =

“Exactly one of the ‘cron’, ‘fixedDelay(String)’, or ‘fixedRate(String)’ attributes is required”;		Set<ScheduledTask> tasks = new LinkedHashSet<>(4);
	// Determine initial delay
	long initialDelay = convertToMillis(scheduled.initialDelay(), scheduled.timeUnit());
	String initialDelayString = scheduled.initialDelayString();
	if (StringUtils.hasText(initialDelayString)) {
		Assert.isTrue(initialDelay < 0, "Specify 'initialDelay' or 'initialDelayString', not both");
		if (this.embeddedValueResolver != null) {
			initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);
		}
		if (StringUtils.hasLength(initialDelayString)) {
			try {
				initialDelay = convertToMillis(initialDelayString, scheduled.timeUnit());
			}
			catch (RuntimeException ex) {
				throw new IllegalArgumentException(
						"Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
			}
		}
	}

	// Check cron expression
	String cron = scheduled.cron();
	if (StringUtils.hasText(cron)) {
		String zone = scheduled.zone();
		if (this.embeddedValueResolver != null) {
			cron = this.embeddedValueResolver.resolveStringValue(cron);
			zone = this.embeddedValueResolver.resolveStringValue(zone);
		}
		if (StringUtils.hasLength(cron)) {
			Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");
			processedSchedule = true;
			if (!Scheduled.CRON_DISABLED.equals(cron)) {
				TimeZone timeZone;
				if (StringUtils.hasText(zone)) {
					timeZone = StringUtils.parseTimeZoneString(zone);
				}
				else {
					timeZone = TimeZone.getDefault();
				}
				tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, new CronTrigger(cron, timeZone))));
			}
		}
	}

	// At this point we don't need to differentiate between initial delay set or not anymore
	if (initialDelay < 0) {
		initialDelay = 0;
	}

	// Check fixed delay
	long fixedDelay = convertToMillis(scheduled.fixedDelay(), scheduled.timeUnit());
	if (fixedDelay >= 0) {
		Assert.isTrue(!processedSchedule, errorMessage);
		processedSchedule = true;
		tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
	}

	String fixedDelayString = scheduled.fixedDelayString();
	if (StringUtils.hasText(fixedDelayString)) {
		if (this.embeddedValueResolver != null) {
			fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);
		}
		if (StringUtils.hasLength(fixedDelayString)) {
			Assert.isTrue(!processedSchedule, errorMessage);
			processedSchedule = true;
			try {
				fixedDelay = convertToMillis(fixedDelayString, scheduled.timeUnit());
			}
			catch (RuntimeException ex) {
				throw new IllegalArgumentException(
						"Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");
			}
			tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
		}
	}

	// Check fixed rate
	long fixedRate = convertToMillis(scheduled.fixedRate(), scheduled.timeUnit());
	if (fixedRate >= 0) {
		Assert.isTrue(!processedSchedule, errorMessage);
		processedSchedule = true;
		tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
	}
	String fixedRateString = scheduled.fixedRateString();
	if (StringUtils.hasText(fixedRateString)) {
		if (this.embeddedValueResolver != null) {
			fixedRateString = this.embeddedValueResolver.resolveStringValue(fixedRateString);
		}
		if (StringUtils.hasLength(fixedRateString)) {
			Assert.isTrue(!processedSchedule, errorMessage);
			processedSchedule = true;
			try {
				fixedRate = convertToMillis(fixedRateString, scheduled.timeUnit());
			}
			catch (RuntimeException ex) {
				throw new IllegalArgumentException(
						"Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");
			}
			tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
		}
	}

	// Check whether we had any attribute set
	Assert.isTrue(processedSchedule, errorMessage);

	// Finally register the scheduled tasks
	synchronized (this.scheduledTasks) {
		Set<ScheduledTask> regTasks = this.scheduledTasks.computeIfAbsent(bean, key -> new LinkedHashSet<>(4));
		regTasks.addAll(tasks);
	}
}
catch (IllegalArgumentException ex) {
	throw new IllegalStateException(
			"Encountered invalid @Scheduled method '" + method.getName() + "': " + ex.getMessage());
}
}</code></pre><p style="">通过bean的后置处理器，ScheduledAnnotationBeanPostProcessor完成了所有注解方法的扫描工作。而通过监听ContextRefreshedEvent事件，会在所有bean初始化完成之后，进行调度任务的最后阶段工作。完成整个调度任务注册流程。</p><blockquote><p style="">ContextRefreshedEvent会在ApplicationContext初始化完成之后或者刷新的时候发起。</p></blockquote><pre><code class="language-java">@Override

public void onApplicationEvent(ContextRefreshedEvent event) {

if (event.getApplicationContext() == this.applicationContext) {

// Running in an ApplicationContext -&gt; register tasks this late…

// giving other ContextRefreshedEvent listeners a chance to perform

// their work at the same time (e.g. Spring Batch’s job registration).

finishRegistration();

}

}</code></pre><p style="">可以看到主要是调用了finishRegistration方法，而finishRegistration方法的主体逻辑是调用registrar的afterPropertiesSet方法。</p><pre><code>private void finishRegistration() {

if (this.scheduler != null) {

this.registrar.setScheduler(this.scheduler);

}

this.registrar.afterPropertiesSet();
}

ScheduledTaskRegistrar

前面提到了多次的registrar，就是ScheduledTaskRegistrar的实例。

ScheduledTaskRegistrar有这些成员属性：四种类型任务的集合、我们熟悉的ScheduledExecutorService。还有没出现过的TaskScheduler。

@Override

public void afterPropertiesSet() {

scheduleTasks();

}/**
 * Schedule all registered tasks against the underlying
 * {@linkplain #setTaskScheduler(TaskScheduler) task scheduler}.
 */
@SuppressWarnings("deprecation")
protected void scheduleTasks() {
	if (this.taskScheduler == null) {
		this.localExecutor = Executors.newSingleThreadScheduledExecutor();
		this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
	}
	if (this.triggerTasks != null) {
		for (TriggerTask task : this.triggerTasks) {
			addScheduledTask(scheduleTriggerTask(task));
		}
	}
	if (this.cronTasks != null) {
		for (CronTask task : this.cronTasks) {
			addScheduledTask(scheduleCronTask(task));
		}
	}
	if (this.fixedRateTasks != null) {
		for (IntervalTask task : this.fixedRateTasks) {
			addScheduledTask(scheduleFixedRateTask(task));
		}
	}
	if (this.fixedDelayTasks != null) {
		for (IntervalTask task : this.fixedDelayTasks) {
			addScheduledTask(scheduleFixedDelayTask(task));
		}
	}
}</code></pre><p style="">可以看到，afterPropertiesSet的主体就是调用scheduleTasks。而scheduleTasks方法的核心逻辑是：</p><ol><li><p style="">如果taskScheduler为空，则初始化localExecutor和taskScheduler。</p></li><li><p style="">对四种不同类型的任务，循环加入调度。多的一种TriggerTask，是自定义任务。</p></li></ol><p style="">我们先来看下比较熟悉的scheduleFixedRateTask和scheduleFixedDelayTask。这两个逻辑基本一致，所以我们以scheduleFixedRateTask为例。</p><pre><code class="language-java">@Deprecated
@Nullable
public ScheduledTask scheduleFixedRateTask(IntervalTask task) {
	FixedRateTask taskToUse = (task instanceof FixedRateTask ? (FixedRateTask) task :
			new FixedRateTask(task.getRunnable(), task.getInterval(), task.getInitialDelay()));
	return scheduleFixedRateTask(taskToUse);
}
/**Schedule the specified fixed-rate task, either right away if possibleor on initialization of the scheduler.@return a handle to the scheduled task, allowing to cancel it(or {@code null} if processing a previously registered task)@since 5.0.2

*/

@Nullable

public ScheduledTask scheduleFixedRateTask(FixedRateTask task) {

ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);

if (this.taskScheduler != null) {

if (task.getInitialDelay() &gt; 0) {

Date startTime = new Date(this.taskScheduler.getClock().millis() + task.getInitialDelay());

scheduledTask.future =

this.taskScheduler.scheduleAtFixedRate(task.getRunnable(), startTime, task.getInterval());

}

else {

scheduledTask.future =

this.taskScheduler.scheduleAtFixedRate(task.getRunnable(), task.getInterval());

}

}

else {

addFixedRateTask(task);

this.unresolvedTasks.put(task, scheduledTask);

}

return (newTask ? scheduledTask : null);

}</code></pre><ol><li><p style="">对任务进行封装，然后调用同名方法scheduleFixedRateTask</p></li><li><p style="">然后根据任务的不同配置，是否有InitialDelay，调用taskScheduler的不同方法去提交任务。</p></li></ol><h4 style="" id="threadpooltaskscheduler">ThreadPoolTaskScheduler</h4><p style="">ThreadPoolTaskScheduler是TaskScheduler的一个实现类。是Spring Scheduling对JDK ScheduledExecutorService的又一层封装。</p><p style="">上面我们看到，在ScheduledTaskRegistrar中提交固定频率的任务后，最终会调用this.taskScheduler.scheduleAtFixedRate方法。而在taskScheduler.scheduleAtFixedRate中，又最终会调用ScheduledExecutorService的scheduleAtFixedRate方法。</p><pre><code class="language-java">@Override

public ScheduledFuture&lt;?&gt; scheduleAtFixedRate(Runnable task, Date startTime, long period) {

ScheduledExecutorService executor = getScheduledExecutor();

long initialDelay = startTime.getTime() - this.clock.millis();

try {

return executor.scheduleAtFixedRate(errorHandlingTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);

}

catch (RejectedExecutionException ex) {

throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);

}

}</code></pre><h4 style="" id="cron%E8%A1%A8%E8%BE%BE%E5%BC%8F%E4%BB%BB%E5%8A%A1%E6%98%AF%E5%A6%82%E4%BD%95%E5%AE%9E%E7%8E%B0%E7%9A%84">Cron表达式任务是如何实现的</h4><p style="">我们知道Spring Scheduling是对ScheduledExecutorService的进一步封装。ScheduledExecutorService只支持固定延迟、固定频率、单次任务这三种任务，而Spring Scheduling还支持cron表达式任务，这个是怎么实现的呢？</p><p style="">我们要先回到ScheduledTaskRegistrar的scheduleCronTask方法。</p><pre><code class="language-java">@Nullable

public ScheduledTask scheduleCronTask(CronTask task) {

ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);

boolean newTask = false;

if (scheduledTask == null) {

scheduledTask = new ScheduledTask(task);

newTask = true;

}

if (this.taskScheduler != null) {

scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());

}

else {

addCronTask(task);

this.unresolvedTasks.put(task, scheduledTask);

}

return (newTask ? scheduledTask : null);

}</code></pre><p style="">调用的是taskScheduler的schedule方法。入参是Runnable实例和1个Trigger。</p><blockquote><p style="">Trigger是触发器，一般是与任务关联的，用于计算任务的下次执行时间。而CronTask的触发器就是CronTrigger。</p></blockquote><pre><code class="language-java">@Override

@Nullable

public ScheduledFuture&lt;?&gt; schedule(Runnable task, Trigger trigger) {

ScheduledExecutorService executor = getScheduledExecutor();

try {

ErrorHandler errorHandler = this.errorHandler;

if (errorHandler == null) {

errorHandler = TaskUtils.getDefaultErrorHandler(true);

}

return new ReschedulingRunnable(task, trigger, this.clock, executor, errorHandler).schedule();

}

catch (RejectedExecutionException ex) {

throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);

}

}</code></pre><p style="">仍然是通过ScheduledExecutorService来实现的，但是在提交之前，有些额外的处理：</p><ol><li><p style="">Runnable任务又被封装了一层，类型是ReschedulingRunnable。Rescheduling是“重新调度“的意思。</p></li><li><p style="">调用新生成的ReschedulingRunnable实例的schedule方法。</p></li></ol><pre><code class="language-java">public ReschedulingRunnable(Runnable delegate, Trigger trigger, Clock clock,

ScheduledExecutorService executor, ErrorHandler errorHandler) {super(delegate, errorHandler);

this.trigger = trigger;

this.triggerContext = new SimpleTriggerContext(clock);

this.executor = executor;

}@Nullable

public ScheduledFuture&lt;?&gt; schedule() {

synchronized (this.triggerContextMonitor) {

this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);

if (this.scheduledExecutionTime == null) {

return null;

}

long initialDelay = this.scheduledExecutionTime.getTime() - this.triggerContext.getClock().millis();

this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);

return this;

}

}
@Override

public void run() {

Date actualExecutionTime = new Date(this.triggerContext.getClock().millis());

super.run();

Date completionTime = new Date(this.triggerContext.getClock().millis());

synchronized (this.triggerContextMonitor) {

Assert.state(this.scheduledExecutionTime != null, “No scheduled execution”);

this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);

if (!obtainCurrentFuture().isCancelled()) {

schedule();

}

}

}

ReschedulingRunnable的构造方法入参包含ScheduledExecutorService的实例。其核心逻辑在于schedule方法和run方法。

schedule方法通过trigger计算出任务的执行时间，然后通过ScheduledExecutorService提交任务，一定要注意的是这里提交的是一次性任务。schedule方法并没有period参数。

那cron表达式式任务是如何实现周期执行的，重点在于run方法。ReschedulingRunnable继承了父类DelegatingErrorHandlingRunnable。我们的业务逻辑Runnable实例在ReschedulingRunnable的构造方法中，被传入了父类的属性中。所以在ReschedulingRunnable的run方法中，super.run是我们的业务逻辑。在业务逻辑后面，ReschedulingRunnable包装了额外的逻辑，就是再次调用schedule方法，计算下次任务的时间并且重新提交。

总结

本文从使用和源码实现层面介绍了Java中多种定时任务的实现方案。为了不离题太远和控制篇幅，更能体现本文的主要脉络。其实省略了一些技术细节。比如：

1. Timer、ScheduledExecutorService、Spring Scheduling中涉及的线程安全问题的讨论，锁的应用等
2. 阻塞队列
3. 异步编程相关的知识，如Future。

大家可以自行补充，后面我可能也会补充这些技术点的文章。

参考

不用任何框架，Java 就能实现定时任务的 3 种方法！

[重要提示]

所有博客内容，在我的个人博客网站可见，欢迎访问： TwoFish
本文由博客一文多发平台 OpenWrite 发布！