一、摘要
在很多业务的系统中，我们常常需要定时的执行一些任务，例如定时发短信、定时变更数据、定时发起促销活动等等。
在上篇文章中，我们简单的介绍了定时任务的使用方式，不同的架构对应的解决方案也有所不同，总结起来主要分单机和分布式两大类，本文会重点分析下单机的定时任务实现原理以及优缺点，分布式框架的实现原理会在后续文章中进行分析。
从单机角度，定时任务实现主要有以下 3 种方案：
- while + sleep 组合
- 最小堆实现
- 时间轮实现
二、while+sleep组合
while+sleep 方案，简单的说，就是定义一个线程，然后 while 循环，通过 sleep 延迟时间来达到周期性调度任务。
简单示例如下：
public static void main(String[] args) { final long timeInterval = 5000; new Thread(new Runnable() { @Override public void run() { while (true) { System.out.println(Thread.currentThread().getName() + "每隔5秒执行一次"); try { Thread.sleep(timeInterval); } catch (InterruptedException e) { e.printStackTrace(); } } } }).start(); }
实现上非常简单，如果我们想在创建一个每隔3秒钟执行一次任务，怎么办呢？
同样的，也可以在创建一个线程，然后间隔性的调度方法；但是如果创建了大量这种类型的线程，这个时候会发现大量的定时任务线程在调度切换时性能消耗会非常大，而且整体效率低！
面对这种在情况，大佬们也想到了，于是想出了用一个线程将所有的定时任务存起来，事先排好序，按照一定的规则来调度，这样不就可以极大的减少每个线程的切换消耗吗？
正因此，JDK 中的 Timer 定时器由此诞生了！
三、最小堆实现
所谓最小堆方案，正如我们上面所说的，每当有新任务加入的时候，会把需要即将要执行的任务排到前面，同时会有一个线程不断的轮询判断，如果当前某个任务已经到达执行时间点，就会立即执行，具体实现代表就是 JDK 中的 Timer 定时器！
3.1、Timer
首先我们来一个简单的 Timer 定时器例子
public static void main(String[] args) { Timer timer = new Timer(); //每隔1秒调用一次 timer.schedule(new TimerTask() { @Override public void run() { System.out.println("test1"); } }, 1000, 1000); //每隔3秒调用一次 timer.schedule(new TimerTask() { @Override public void run() { System.out.println("test2"); } }, 3000, 3000); }
实现上，好像跟我们上面介绍的 while+sleep 方案差不多，同样也是起一个TimerTask线程任务，只不过共用一个Timer调度器。
下面我们一起来打开源码看看里面到底有些啥！
- 进入Timer.schedule()方法
从方法上可以看出，这里主要做参数验证，其中
TimerTask是一个线程任务，delay表示延迟多久执行（单位毫秒），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); }
- 接着看
sched()方法
这步操作中，可以很清晰的看到，在同步代码块里，会将
task对象加入到queue
private void sched(TimerTask task, long time, long period) { if (time < 0) throw new IllegalArgumentException("Illegal execution time."); // Constrain value of period sufficiently to prevent numeric // overflow while still being effectively infinitely large. if (Math.abs(period) > (Long.MAX_VALUE >> 1)) period >>= 1; synchronized(queue) { if (!thread.newTasksMayBeScheduled) throw new IllegalStateException("Timer already cancelled."); synchronized(task.lock) { if (task.state != TimerTask.VIRGIN) throw new IllegalStateException( "Task already scheduled or cancelled"); task.nextExecutionTime = time; task.period = period; task.state = TimerTask.SCHEDULED; } queue.add(task); if (queue.getMin() == task) queue.notify(); } }
- 我们继续来看
queue对象
任务会将入到
TaskQueue队列中，同时在Timer初始化阶段会将TaskQueue作为参数传入到TimerThread线程中，并且起到线程
public class Timer { private final TaskQueue queue = new TaskQueue(); private final TimerThread thread = new TimerThread(queue); public Timer() { this("Timer-" + serialNumber()); } public Timer(String name) { thread.setName(name); thread.start(); } //... }
- 而
TaskQueue其实是一个最小堆的数据实体类，源码如下
每当有新元素加入的时候，会对原来的数组进行重排，会将即将要执行的任务排在数组的前面
class TaskQueue { private TimerTask[] queue = new TimerTask[128]; private int size = 0; void add(TimerTask task) { // Grow backing store if necessary if (size + 1 == queue.length) queue = Arrays.copyOf(queue, 2*queue.length); queue[++size] = task; fixUp(size); } private void fixUp(int k) { while (k > 1) { int j = k >> 1; if (queue[j].nextExecutionTime <= queue[k].nextExecutionTime) break; TimerTask tmp = queue[j]; queue[j] = queue[k]; queue[k] = tmp; k = j; } } //.... }
- 最后我们来看看
TimerThread
TimerThread其实就是一个任务调度线程，首先从TaskQueue里面获取排在最前面的任务，然后判断它是否到达任务执行时间点，如果已到达，就会立刻执行任务
class TimerThread extends Thread { boolean newTasksMayBeScheduled = true; 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 } } } /** * The main timer loop. (See class comment.) */ 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 evt 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) { } } } }
总结这个利用最小堆实现的方案，相比 while + sleep 方案，多了一个线程来管理所有的任务，优点就是减少了线程之间的性能开销，提升了执行效率；但是同样也带来的了一些缺点，整体的新加任务写入效率变成了 O(log(n))。
同时，细心的发现，这个方案还有以下几个缺点：
- 串行阻塞：调度线程只有一个，长任务会阻塞短任务的执行，例如，A任务跑了一分钟，B任务至少需要等1分钟才能跑
- 容错能力差：没有异常处理能力，一旦一个任务执行故障，后续任务都无法执行
3.2、ScheduledThreadPoolExecutor
鉴于 Timer 的上述缺陷，从 Java 5 开始，推出了基于线程池设计的 ScheduledThreadPoolExecutor 。
其设计思想是，每一个被调度的任务都会由线程池来管理执行，因此任务是并发执行的，相互之间不会受到干扰。需要注意的是，只有当任务的执行时间到来时，ScheduledThreadPoolExecutor 才会真正启动一个线程，其余时间 ScheduledThreadPoolExecutor 都是在轮询任务的状态。
