ScheduleThreadPoolExecutor应用源码

4.1 ScheduleThreadPoolExecutor介绍

从名字上就可以看出，当前线程池是用于执行定时任务的线程池。
Java比较早的定时任务工具是Timer类。但是Timer问题很多，串行的，不靠谱，会影响到其他的任务执行。
其实除了Timer以及ScheduleThreadPoolExecutor之外，正常在企业中一般会采用Quartz或者是SpringBoot提供的Schedule的方式去实现定时任务的功能。
ScheduleThreadPoolExecutor支持延迟执行以及周期性执行的功能。

4.2 ScheduleThreadPoolExecutor应用

定时任务线程池的有参构造

public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}

发现ScheduleThreadPoolExecutor在构建时，直接调用了父类的构造方法
ScheduleThreadPoolExecutor的父类就是ThreadPoolExecutor
首先ScheduleThreadPoolExecutor最多允许设置3个参数：
● 核心线程数
● 线程工厂
● 拒绝策略
首先没有设置阻塞队列，以及最大线程数和空闲时间以及单位
阻塞队列设置的是DelayedWorkQueue，其实本质就是DelayQueue，一个延迟队列。DelayQueue是一个无界队列。所以最大线
程数以及非核心线程的空闲时间是不需要设置的。
代码落地使用

public static void main(String[] args) {
//1. 构建定时任务线程池
ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(
5,
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
return t;
}
},
new ThreadPoolExecutor.AbortPolicy());
//2. 应用ScheduledThreadPoolExecutor
// 跟直接执行线程池的execute没啥区别
pool.execute(() -> {
System.out.println("execute");
});
// 指定延迟时间执行
System.out.println(System.currentTimeMillis());
pool.schedule(() -> {
System.out.println("schedule");
System.out.println(System.currentTimeMillis());
},2, TimeUnit.SECONDS);
// 指定第一次的延迟时间，并且确认后期的周期执行时间，周期时间是在任务开始时就计算
// 周期性执行就是将执行完毕的任务再次社会好延迟时间，并且重新扔到阻塞队列
// 计算的周期执行，也是在原有的时间上做累加，不关注任务的执行时长。
System.out.println(System.currentTimeMillis());
pool.scheduleAtFixedRate(() -> {
System.out.println("scheduleAtFixedRate");
System.out.println(System.currentTimeMillis());
},2,3,TimeUnit.SECONDS);
// // 指定第一次的延迟时间，并且确认后期的周期执行时间，周期时间是在任务结束后再计算下次的延迟时间
System.out.println(System.currentTimeMillis());pool.scheduleWithFixedDelay(() -> {
System.out.println("scheduleWithFixedDelay");
System.out.println(System.currentTimeMillis());
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
},2,3,TimeUnit.SECONDS);
}

4.3 ScheduleThreadPoolExecutor源码剖析

4.3.1 核心属性

后面的方法业务流程会涉及到这些属性。

// 这里是针对任务取消时的一些业务判断会用到的标记
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
private volatile boolean removeOnCancel = false;
// 计数器，如果两个任务的执行时间节点一模一样，根据这个序列来判断谁先执行
private static final AtomicLong sequencer = new AtomicLong();
// 这个方法是获取当前系统时间的毫秒值
final long now() {
return System.nanoTime();
}
// 内部类。核心类之一。
private class ScheduledFutureTask<V>
extends FutureTask<V> implements RunnableScheduledFuture<V> {
// 全局唯一的序列，如果两个任务时间一直，基于当前属性判断
private final long sequenceNumber;
// 任务执行的时间，单位纳秒
private long time;
/**
* period == 0：执行一次的延迟任务
* period > 0：代表是At
* period < 0：代表是With
*/
private final long period;
// 周期性执行时，需要将任务重新扔回阻塞队列，基础当前属性拿到任务，方便扔回阻塞队列
RunnableScheduledFuture<V> outerTask = this;
/**
* 构建schedule方法的任务*/
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
/**
* 构建At和With任务的有参构造
*/
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement();
}
}
// 内部类。核心类之一。
static class DelayedWorkQueue extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {
// 这个类就是DelayQueue，不用过分关注，如果没看过，看阻塞队列中的优先级队列和延迟队列

4.3.2 schedule方法

execute方法也是调用的schedule方法，只不过传入的延迟时间是0纳秒
schedule方法就是将任务和延迟时间封装到一起，并且将任务扔到阻塞队列中，再去创建工作线程去take阻塞队列。

// 延迟任务执行的方法。
// command：任务
// delay：延迟时间
// unit：延迟时间的单位
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
// 健壮性校验。
if (command == null || unit == null)
throw new NullPointerException();
// 将任务和延迟时间封装到一起，最终组成ScheduledFutureTask
// 要分成三个方法去看
// triggerTime：计算延迟时间。最终返回的是当前系统时间 + 延迟时间
// triggerTime就是将延迟时间转换为纳秒，并且+当前系统时间，再做一些健壮性校验
// ScheduledFutureTask有参构造：将任务以及延迟时间封装到一起，并且设置任务执行的方式
// decorateTask：当前方式是让用户基于自身情况可以动态修改任务的一个扩展口
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
// 任务封装好，执行delayedExecute方法，去执行任务
delayedExecute(t);
// 返回FutureTaskreturn t;
}
// triggerTime做的事情
// 外部方法，对延迟时间做校验，如果小于0，就直接设置为0
// 并且转换为纳秒单位
private long triggerTime(long delay, TimeUnit unit) {
return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
// 将延迟时间+当前系统时间
// 后面的校验是为了避免延迟时间超过Long的取值范围
long triggerTime(long delay) {
return now() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}
// ScheduledFutureTask有参构造
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
// time就是任务要执行的时间
this.time = ns;
// period,为0，代表任务是延迟执行，不是周期执行
this.period = 0;
// 基于AtmoicLong生成的序列
this.sequenceNumber = sequencer.getAndIncrement();
}
// delayedExecute 执行延迟任务的操作
private void delayedExecute(RunnableScheduledFuture<?> task) {
// 查看当前线程池是否还是RUNNING状态，如果不是RUNNING，进到if
if (isShutdown())
// 不是RUNNING。
// 执行拒绝策略。
reject(task);
else {
// 线程池状态是RUNNING
// 直接让任务扔到延迟的阻塞队列中
super.getQueue().add(task);
// DCL的操作，再次查看线程池状态
// 如果线程池在添加任务到阻塞队列后，状态不是RUNNING
if (isShutdown() &&
// task.isPeriodic()：现在反回的是false，因为任务是延迟执行，不是周期执行
// 默认情况，延迟队列中的延迟任务，可以执行
!canRunInCurrentRunState(task.isPeriodic()) &&
// 从阻塞队列中移除任务。
remove(task))
task.cancel(false);
else
// 线程池状态正常，任务可以执行
ensurePrestart();
}
}
// 线程池状态不为RUNNING，查看任务是否可以执行
// 延迟执行：periodic==false
// 周期执行：periodic==true
// continueExistingPeriodicTasksAfterShutdown：周期执行任务，默认为false
// executeExistingDelayedTasksAfterShutdown：延迟执行任务，默认为true
boolean canRunInCurrentRunState(boolean periodic) {
return isRunningOrShutdown(periodic ?
continueExistingPeriodicTasksAfterShutdown :
executeExistingDelayedTasksAfterShutdown);
}
// 当前情况，shutdownOK为true
final boolean isRunningOrShutdown(boolean shutdownOK) {
int rs = runStateOf(ctl.get());
// 如果状态是RUNNING，正常可以执行，返回true
// 如果状态是SHUTDOWN，根据shutdownOK来决定
return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
}
// 任务可以正常执行后，做的操作
void ensurePrestart() {
// 拿到工作线程个数
int wc = workerCountOf(ctl.get());
// 如果工作线程个数小于核心线程数
if (wc < corePoolSize)
// 添加核心线程去处理阻塞队列中的任务
addWorker(null, true);
else if (wc == 0)// 如果工作线程数为0，核心线程数也为0，这是添加一个非核心线程去处理阻塞队列任务
addWorker(null, false);
}

4.3.3 At和With方法&任务的run方法

这两个方法在源码层面上的第一个区别，就是在计算周期时间时，需要将这个值传递给period，基于正负数在区别At和With
所以查看一个方法就ok，查看At方法

// At方法，
// command：任务
// initialDelay：第一次执行的延迟时间
// period：任务的周期执行时间
// unit：上面两个时间的单位
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit) {
// 健壮性校验
if (command == null || unit == null)
throw new NullPointerException();
// 周期时间不能小于等于0.
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);
// 周期性任务，需要在任务执行完毕后，重新扔会到阻塞队列，为了方便拿任务，将任务设置到outerTask成员变量中
sft.outerTask = t;
// 和schedule方法一样的方式
// 如果任务刚刚扔到阻塞队列，线程池状态变为SHUTDOWN，默认情况，当前任务不执行
delayedExecute(t);
return t;
}
// 延迟任务以及周期任务在执行时，都会调用当前任务的run方法。
public void run() {
// periodic == false：一次性延迟任务
// periodic == true：周期任务
boolean periodic = isPeriodic();
// 任务执行前，会再次判断状态，能否执行任务
if (!canRunInCurrentRunState(periodic))
cancel(false);
// 判断是周期执行还是一次性任务
else if (!periodic)
// 一次性任务，让工作线程直接执行command的逻辑
ScheduledFutureTask.super.run();// 到这个else if，说明任务是周期执行
else if (ScheduledFutureTask.super.runAndReset()) {
// 设置下次任务执行的时间
setNextRunTime();
// 将任务重新扔回线程池做处理
reExecutePeriodic(outerTask);
}
}
// 设置下次任务执行的时间
private void setNextRunTime() {
// 拿到period值，正数：At，负数：With
long p = period;
if (p > 0)
// 拿着之前的执行时间，直接追加上周期时间
time += p;
else
// 如果走到else，代表任务是With方式，这种方式要重新计算延迟时间
// 拿到当前系统时间，追加上延迟时间，
time = triggerTime(-p);
}
// 将任务重新扔回线程池做处理
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
// 如果状态ok，可以执行
if (canRunInCurrentRunState(true)) {
// 将任务扔到延迟队列
super.getQueue().add(task);
// DCL，判断线程池状态if (!canRunInCurrentRunState(true) && remove(task))
task.cancel(false);
else
// 添加工作线程
ensurePrestart();
}
}