ScheduledThreadPoolExecutor 分析

一、ScheduledThreadPoolExecutor介绍

从名字就可以看出,当前线程池是用来执行定时任务的线程池

Java早期的定时任务操作是由Timer类提供的,但是Timer有很多问题
他是串行执行的不能像线程池这样搞个子线程去执行、而且可能会影响到其他的定时任务执行

ScheduledThreadPoolExecutor支持延迟执行 以及 周期执行的功能
周期性: 这里的周期性就是将执行完的任务 重新计算下次需要执行的时间,再次把任务扔到阻塞队列中

二、ScheduledThreadPoolExecutor应用

2.1 ScheduledThreadPoolExecutor 构造器

// ScheduledThreadPoolExecutor继承了ThreadPoolExecutor 因此其执行的流程还是ThreadPoolExecutor来的
public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    
}

/**
 * 调用父类的构造器 
 * 
 * ScheduledThreadPoolExecutor 允许我们最多设置三个参数
 *  - 核心线程数
 *  - 线程工厂
 *  - 拒绝策略
 *  
 *  不需要设置阻塞队列 , 默认提供给我们 DelayedWorkQueue , 类似优先级队列 是一个无界队列 
 *             当核心线程数满了之后 就添加到这个无界队列中 也就没有机会创建非核心线程了,因此不需要设置最大线程数量默认Integer.MAX_VALUE
 *  
 * 
 */
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

2.2 ScheduledThreadPoolExecutor 应用代码

public class TestScheduledThreadPoolExecutor {

    public static void main(String[] args) {
        // 1. 构建ScheduledThreadPoolExecutor
        ScheduledThreadPoolExecutor poolExecutor = new ScheduledThreadPoolExecutor(
                5,
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        return thread;
                    }
                },
                new ThreadPoolExecutor.AbortPolicy()
        );

        // 2.应用ScheduledThreadPoolExecutor

        // 正常的execute方法
        poolExecutor.execute(() -> {
            System.out.println("execute");
        });

        // 延迟两秒执行
        poolExecutor.schedule(() -> {
            System.out.println("schedule");
        },3 , TimeUnit.SECONDS);

        // 第一次延时三秒执行这个任务 后边每两秒执行一次
        // 开始执行时就确认了下次要执行的时间
        poolExecutor.scheduleAtFixedRate(() -> {
            System.out.println("scheduleAtFixedRate");
        },3,2,TimeUnit.SECONDS);

        // 和上边的区别就是 这个方式使用定时任务 是在任务执行完成后才计算的下次执行任务的时间的
        poolExecutor.scheduleWithFixedDelay(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("scheduleWithFixedDelay");
        },3,2,TimeUnit.SECONDS);
        
    }
}

三、ScheduledThreadPoolExecutor源码分析

3.1 核心属性

// 这三个标记 是用于任务取消操作相关的判断
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
private volatile boolean removeOnCancel = false;
// 计数器,如果两个任务的执行时间节点一样,那么就需要根据这个序列来判断谁先谁后
private static final AtomicLong sequencer = new AtomicLong();

3.2 核心内部类 (和前边一样其实就是让任务具有延时性的类 实现Delayed接口)

想要实现延时性 那么就需要实现Delayed接口定义好其 比较规则和延时时间到期的规则 然后才能使用延时队列进行添加任务

ScheduledFutureTask类结构图

private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {
    // 当前任务的 全局唯一的 序列值
    // 如果两个任务时间一致 就根据当前属性判断
    private final long sequenceNumber;
    // 任务执行的时间,单位是纳秒
    private long time;
    /**
     * period == 0 : 说明是只执行一次的任务 不是周期性任务
     * period >  0 : 说明是执行 scheduleAtFixedRate , 以固定周期执行的任务
     * period <  0 : 说明是执行 scheduleWithFixedDelay , 加上任务的执行时间后再执行任务
     */
    private final long period;
    // 周期性执行时 需要将任务重新扔到阻塞队列中,基于当前属性 拿到这个任务
    RunnableScheduledFuture<V> outerTask = this;
    
    int heapIndex;

    /**
     * period == 0 构建的是schedule方法执行的任务
     */
    ScheduledFutureTask(Runnable r, V result, long ns) {
        super(r, result);
        this.time = ns;
        this.period = 0;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

    /**
     * 构建周期性的任务
     */
    ScheduledFutureTask(Runnable r, V result, long ns, long period) {
        super(r, result);
        this.time = ns;
        this.period = period;
        this.sequenceNumber = sequencer.getAndIncrement();
    }
}

3.3 schedule 方法分析

execute 也是调用的schedule方法,只不过延时时间是0纳秒
将任务和时间封装在一起组成一个具有延时性的对象(ScheduleFutureTask) 然后扔到阻塞队列中进行执行

3.3.1 schedule 方法

/** 
 * 延迟任务执行的方法
 * command : 要执行的任务
 * delay : 延迟时间
 * unit  : 延时时间的单位
 */
public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    
    // 将任务和延时时间封装到一起(ScheduledFutureTask延时任务对象)
    // triggerTime方法: 将我们传入的延时时间和单位转换为该任务要执行的时间 
    //                 当前系统时间+设置的延时时间的结果 就是任务要执行的时间
    // ScheduledFutureTask有参构造: 将任务 延时时间封装在一起组成一个延时任务对象
    // decorateTask: 仅仅是进入了一下方法,利用多态返回父类类型
    RunnableScheduledFuture<?> t = 
        decorateTask(
                command,
                new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit))
        );
    // 封装好延时任务后 执行这个任务
    delayedExecute(t);
    // 返回了FutureTask对象
    return t;
}

3.3.2 triggerTime 方法

private long triggerTime(long delay, TimeUnit unit) {
    // 对延时时间做校验
    // 如果延时时间 < 0 直接设置为0 , 并把延时时间转换为纳秒 , 延时任务的时间<0没有意义
    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);
    this.time = ns;
    // period == 0 说明是延时执行而不是周期执行
    this.period = 0;
    // 基于AtomicLong生成的唯一序列
    this.sequenceNumber = sequencer.getAndIncrement();
}

3.3.3 decorateTask

// 这个方法直接返回了传入的延时任务 task
// 这个方法用户可根据自己的要求 拓展该方法 对任务进行修改
protected <V> RunnableScheduledFuture<V> decorateTask(
    Runnable runnable, RunnableScheduledFuture<V> task) {
    return task;
}

3.3.4 delayedExecute

// 执行延时任务
private void delayedExecute(RunnableScheduledFuture<?> task) {
    // 查看当前线程池状态 是否是RUNNING
    if (isShutdown())
        // 不是RUNNING状态不接受新任务 , 执行拒绝策略
        reject(task);
    else {
        // 是RUNNING状态
        
        // **这里是直接是把任务放到了延时阻塞队列中**
        // 这里和ThreadPoolExecutor不同的是 不会先创建核心线程 而是先将任务扔到队列中
        super.getQueue().add(task);
        /*
         * 放入任务后再次判断线程池状态
         * 如果加入任务后,线程池状态变为了SHUTDOWN状态,那么此时要针对这个任务来决定还要不要执行
         */
        if (isShutdown() &&
            // task.isPeriodic() 返回true 说明是周期性执行,但是这里periodic==0不是周期执行的 所以这里会返回false , 因此会继续执行下边的判断
            // canRunInCurrentRunState,判断这个周期任务能否执行,方法返回true说明可以执行 再加上! 则不进if逻辑
            // 默认情况下 SHUTDOWN状态下 延时任务可以执行 , 周期任务不能执行
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
            // 线程池状态正常 , 任务可以执行
            ensurePrestart();
    }
}

// 线程池状态是SHUTDOWN时,判断任务是否可以执行的方法
// 延时执行: periodic==false
// 周期执行: periodic==true
// executeExistingDelayedTasksAfterShutdown: 延时任务 在SHUTDOWN状态下默认为true
// continueExistingPeriodicTasksAfterShutdown: 周期任务 在SHUTDOWN状态下默认为false
boolean canRunInCurrentRunState(boolean periodic) {
    return isRunningOrShutdown(periodic ?
                               continueExistingPeriodicTasksAfterShutdown :
                               executeExistingDelayedTasksAfterShutdown);
}
// 属性默认值
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

// shutdownOK : SHUTDOWN下能否执行 这里是true
final boolean isRunningOrShutdown(boolean shutdownOK) {
    int rs = runStateOf(ctl.get());
    // 状态是RUNNING , 则返回true
    // 状态时SHUTDOWN , 最后就根据 shutdownOK 来决定
    return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
}

3.3.4.1 ensurePrestart 正常执行延时任务方法

// 线程池状态正常可以执行延时任务的方法
void ensurePrestart() {
    // 拿到工作线程数量
    int wc = workerCountOf(ctl.get());
    // 数量少于核心线程 添加核心工作线程 , 只不过不带任务 因为是先向队列中加的任务
    if (wc < corePoolSize)
        addWorker(null, true);
    else if (wc == 0)
        // 走到这说明 设置的核心线程数量是0个 工作线程也是0个 , 因此添加一个非核心线程去处理任务
        addWorker(null, false);
}

3.4 scheduleAtFixedRate 方法分析

将任务先延时执行设置的时间单位后执行,然后按照设置的固定时间间隔周期执行

/**
 * 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),
                                      // 这里的时间是 > 0 的
                                      unit.toNanos(period));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    // 周期性任务,需要在任务执行完成之后重新把当前任务扔到阻塞队列,为了拿到任务 将outerTask设置为自己
    sft.outerTask = t;
    // 此时任务是周期任务
    // 如果任务扔到阻塞队列后,线程池状态就变为了SHUTDOWN状态,那么默认情况下周期任务是不能够执行的需要 remove掉此任务
    delayedExecute(t);
    return t;
}

3.5 scheduleWithFixedDelay 方法分析

这个方法和上边at方法的区别 就是传入的周期时间这里转换为了负数,所以核心还是在delayedExecute方法内部来区别这两个方法逻辑

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (delay <= 0)
        throw new IllegalArgumentException();
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      // 和At方法相比只有这里不同
                                      unit.toNanos(-delay));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    delayedExecute(t);
    return t;
}

3.6 延时任务、周期任务，执行具体逻辑

ScheduledThreadPoolExecutor中存放的任务都是其内部类 ScheduledFutureTask类型的任务,所以执行ScheduledFutureTask的run方法

// 延时任务和周期任务的执行方法 , 执行到这个方法说明线程池中的工作线程已经把这个任务从阻塞队列中取出来了(延时时间到了)
public void run() {
    // 判断当前任务是否是周期任务 , 此时this指向的就是正在执行的任务
    boolean periodic = isPeriodic();
    // 任务执行前,再次判断线程池状态 能否执行当前任务
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 可以正常执行
    else if (!periodic)
        // 不是周期任务(一次性延时任务) , 让工作线程执行command的逻辑
        ScheduledFutureTask.super.run();
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 是周期任务
        // 设置下次任务的执行时间
        setNextRunTime();
        // 周期任务,将任务重新扔到阻塞队列中
        reExecutePeriodic(outerTask);
    }
}

3.6.1 isPeriodic

// 延时任务 periodic==0
// 周期任务 periodic!=0
public boolean isPeriodic() {
    return period != 0;
}

3.6.2 setNextRunTime

private void setNextRunTime() {
    // 拿到当前任务的period属性 , period>0:执行At方法 , period<0:执行With方法
    long p = period;
    // 执行At方法
    if (p > 0)
        // At方法设置任务下次执行时间为 本次执行时间+周期时间
        time += p;
    // 反之执行With方法
    else
        // With方法则设置当前任务下次执行时间为 当前时间now() + -p(p设置时为负数此时-p为正数) , 即重新计算任务时间 
        time = triggerTime(-p);
}

3.6.3 reExecutePeriodic

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    // 判断当前周期任务是否可以执行
    if (canRunInCurrentRunState(true)) {
        // 可以执行把 task 扔到阻塞队列中
        super.getQueue().add(task);
        // 再次判断任务能否执行
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            // 添加工作线程 执行任务
            ensurePrestart();
    }
}