Timer类的解析中提到，Timer类的缺陷：
- 单线程
- Timer只有一个任务执行线程，如果某个TimerTask耗时较久，影响其他任务
- Timer不会捕获执行TimerTask时所抛出的异常，由于Timer是单线程，所以一旦出现异常，则线程就会终止，其他任务也得不到执行。
- 系统时间敏感性，由于Timer中使用了currentTimeMillis做参考，所以系统时间变化会使得任务发生变化
所以在JDK1.5后大家就抛弃了Timer，一般使用ScheduledThreadPoolExecutor。
研究完ScheduledExecutor，我突然发现一直困扰我的一个问题——程序中的定时任务是如何实现的？例如，闹钟。ScheduledExecutor正是一个绝佳的例子：
● 任务调度和任务执行解耦，并运行在不同线程上，防止阻塞。任务调度单线程即可满足性能要求，它并不是瓶颈
● 具备延迟时间的任务，在时间的维度上，可以假想它们排成一列，所以你不需要轮询所有任务，只需要关注最早开始的任务
● 等待并不需要轮询，可以等待指定时间唤醒
我们已经看过Timer的延迟执行方案——任务放入最小堆中，任务线程每次取出需要最早执行的任务，并在当前线程中执行。
ScheduledThreadPoolExecutor简单来说，是基于线程池的实现基础上，把原有的任务队列换成DelayQueue。
所以我认为，核心其实在DelayQueue上，它是一种阻塞队列，只有当任务到指定延迟时间时，它才能返回；其次，它也是一种优先队列，基于数组的最小堆实现，按延迟时间排序。
由于取出任务之后，交给线程池执行（由线程池管理子线程执行），所以并不会阻塞调度线程，所以也就没有上述1，2，3的问题。而且它内部使用了nanoTime而不是currentTimeMillis，因而也和系统时间无关（nanoTime的区别是，它是计算从某一个时间的累计时间，这个时间是在启动JVM时任意指定的，例外，它的精度为微纳秒）。
所以它本身不复杂，线程池的实现参考之前的文章。
ScheduledFuture scheduleAtFixedRate(Runnable command,**long **initialDelay,**long **period,TimeUnit unit)，指定任务，初始延迟时间，周期时间，时间单位。注意该方法每一周期启动一次任务，不需要等任务结束。
scheduleWithFixedDelay(Runnable command,**long **initialDelay,**long **delay,TimeUnit unit)，指定任务，初始延迟时间，延迟时间，时间单位。注意该延迟时间指的是前一个周期结束到下一个周期开始时间，换句话说，它需要任务执行完才能开始下一个周期。
ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，复用线程池的功能。实现了ScheduledExecutorService，拓展为调度的线程池服务。
public interface ScheduledExecutorService extends ExecutorService {
public ScheduledFuture schedule(Runnable command,
long delay, TimeUnit unit);
public ScheduledFuture schedule(Callable callable,
long delay, TimeUnit unit);
public ScheduledFuture scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
public ScheduledFuture scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
}
DelayedWorkQueue **extends **AbstractQueue
最后就是队列中放的对象，为ScheduledFutureTask，它的继承结构比较复杂，但本质上就是对FutureTask的拓展，支持了延迟时间的对比查询管理。
ScheduledThreadPoolExecutor的入口，都很简单，核心是队列的阻塞方法——如何实现到时间才返回。
public ScheduledFuture scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (period <= 0L)
throw new IllegalArgumentException();
// 新建了一个延迟任务
ScheduledFutureTask sft =
new ScheduledFutureTask(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period),
sequencer.getAndIncrement());
// decorateTask其实是一个就是隐式的向上转型，转为RunnableScheduledFuture
RunnableScheduledFuture 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 (!canRunInCurrentRunState(task) && remove(task))
task.cancel(false);
else
// 线程池的方法，确认是否添加工作线程
ensurePrestart();
}
}
加入任务队列后，其实剩下都是线程池的工作，它的工作线程会不断地从任务队列中取出任务，区别在于我们的任务队列是DelayedQueue。
所以核心就是理解它如何阻塞线程，让任务到时间再返回，来看take方法。
该Queue持有一个leader，它是当前执行的线程引用。
public RunnableScheduledFuture take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 调度过程加锁，注意lock用于锁定调度
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture first = queue[0];
// 堆为空，等待，available为锁上的condition监视器
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
// 到达延迟时间，任务返回
if (delay <= 0L)
return finishPoll(first);
first = null; // don't retain ref while waiting
// 如果leader不为空，则说明已经有线程占用，available上等待
if (leader != null)
// 除了leader的其他线程无限等待
available.await();
// leader为空，则自己作为leader线程
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// leader等待指定时间
available.awaitNanos(delay);
} finally {
// 执行完，设置leader为null
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 当前任务调度完成，唤醒其他线程调度下一任务
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
我们看到，它的逻辑也不复杂，如果任务达到执行时间，就立刻返回给线程池去执行。每次只有一个线程能做leader，拥有调度当前任务的权限，等待指定延迟时间，其他线程则无限等待，直到被唤醒。
再来说一下leader的作用，这里的leader是为了减少不必要的定时等待，当一个线程成为leader时，它只等待下一个节点的时间间隔，但其它线程无限期等待。 leader线程必须在从take()或poll()返回之前signal其它线程，除非其他线程成为了leader。
然后就是返回任务的时候，queue又做了什么？
其实很简单，将堆顶元素返回，堆顶元素移除，替换为最后一个元素，然后调整堆。
private RunnableScheduledFuture finishPoll(RunnableScheduledFuture f) {
int s = --size;
RunnableScheduledFuture x = queue[s];
queue[s] = null;
if (s != 0)
siftDown(0, x);
setIndex(f, -1);
return f;
}
最后是周期执行如何实现？
答案是，在任务执行完，重新设置下一次的延迟时间，重新入queue调度。
public void run() {
if (!canRunInCurrentRunState(this))
cancel(false);
// 非周期执行，直接执行
else if (!isPeriodic())
super.run();
// 周期执行，执行完重新设置时间
else if (super.runAndReset()) {
// 重新设置时间
setNextRunTime();
// 重新入queue
reExecutePeriodic(outerTask);
}
}
为何不精确
仔细考虑一下整个实现，你会发现：
- 任务执行交给线程池管理
- 调度任务每次只有一个线程能进入，都会等queue中需要最早执行的任务到达执行时间，交给线程池执行。因为一个任务调度很快，所以多个任务先后到达执行时间都会执行，而不像Timer一样，受到前一个任务执行影响
- 由于堆的调整需要时间，线程池的线程资源有限，所以任务也不一定非常精确地按照延迟时间来。
http://concurrent.redspider.group/article/03/20.html
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）