Executors框架二 ScheduledThreadPoolExecutor线程池

Posted 2021-03-16 xieyanke

用途：

ScheduledThreadPoolExecutor(计划任务线程池)主要用于执行一些需要延时操作或者需要重复操作的任务, Spring框架自带计划任务功能

场景一: 延时操作, 提供给客户统计数据功能, 如果高峰执行肯定影响系统运行效率,那么规定只有下午四点以后才能执行统计, 那么就需要用户提交统计任务, 系统待四点以后排队统计用户提交的任务, 用户可以查询任务列表查看此查询任务是否已经完成

场景二: 重复操作, 一般使用在后台管理系统, 提交重复操作的任务, 比如操作人提交一个定时向用户发消息的任务, 此后每隔一段时间向用户发送消息

 

方法解释:

//提交延时计划任务
ScheduledFuture<?> schedule(Runnable command /*任务体 无返回*/, long delay/*延时时间*/, TimeUnit unit/*延时时间单位*/);

//提交延时计划任务
ScheduledFuture<V> schedule(Callable<V> callable/*任务体 有返回*/, long delay/*延时时间*/, TimeUnit unit/*延时时间单位*/);

//提交一个延时后的周期操作任务, 需要注意的是若操作时间长于周期时间, 
//则周期操作会等上一次操作结束之后立即执行, 不会出现上一次未执行完,本次操作到了周期时间开始执行的现象
//并且任务一旦提交后将一直重复执行下去, 除非线程池关闭或者任务抛出异常(抛出异常后续重复操作将不会继续)
ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay /*首次延时时间*/, 
                                        long period /*周期频率*/, TimeUnit unit);

//提交一个延时后的周期操作任务, 需要注意的是若操作时间长于周期时间, 
//则周期操作会等上一次操作结束之后延时执行, 不会出现上一次未执行完,本次操作到了周期时间开始执行的现象
//并且任务一旦提交后将一直重复执行下去, 除非线程池关闭或者任务抛出异常(抛出异常后续重复操作将不会继续)
ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay /*首次延时时间*/,
                                                     long delay /*第一次之后延时时间*/, TimeUnit unit);

使用举例：

ScheduledFuture<?> schedule(Runnable command /*任务体 无返回*/, long delay/*延时时间*/, TimeUnit unit)

public static void main(String[] args/*任务*/) throws Exception {
    long startTime = new Date().getTime();
    ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(10);
    scheduledExecutorService.schedule(()->{
        try {
            Thread.sleep(2000L);
            System.out.println("与启动时相差:"+(new Date().getTime()-startTime)/1000+"秒");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 3000 /*延时3秒后首次执行*/,  TimeUnit.MILLISECONDS);

    scheduledExecutorService.schedule(()->{
        try {
            Thread.sleep(4000L);
            System.out.println("与启动时相差:"+(new Date().getTime()-startTime)/1000+"秒");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 3000 /*延时3秒后首次执行*/,  TimeUnit.MILLISECONDS);
}
结果:
与启动时相差:5秒
与启动时相差:7秒

ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay /*首次延时时间*/, long period /*周期频率*/, TimeUnit unit)

public static void main(String[] args/*任务*/) throws Exception {
    //当前时间毫秒数
    final Long millisecondsNow = new Date().getTime();
    //执行次数
    final AtomicInteger n = new AtomicInteger(0);
    ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(10);
    scheduledExecutorService.scheduleAtFixedRate(()->{
        //计数加1
        n.incrementAndGet();
        //第四次抛出异常 用于测试抛出异常后不再重复操作
        if (n.get()==4){ int a=0; System.out.println(1/a); }

        //测试执行时间
        try {
            System.out.println("第"+n+"次打印 与启动时相差:"+(new Date().getTime()-millisecondsNow)/1000+"秒");
            //睡眠相当于业务执行时间3秒
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 3000 /*延时3秒后首次执行*/, 1000 /*任务间隔1秒*/, TimeUnit.MILLISECONDS);
}
结果:
第1次打印 与启动时相差:3秒
第2次打印 与启动时相差:6秒
第3次打印 与启动时相差:9秒

从上面结果可以看出
1 每次执行结果都是相差3秒, 说明任务间隔时间若小于执行时间, 则每次执行完立即执行下一次任务
2 不会出现前后两次并发的情景出现
3 第四次抛出异常后不再继续执行任务

 

ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay /*首次延时时间*/, long delay /*第一次之后延时时间*/, TimeUnit unit);

public static void main(String[] args/*任务*/) throws Exception {
    //当前时间毫秒数
    final Long millisecondsNow = new Date().getTime();
    //执行次数
    final AtomicInteger n = new AtomicInteger(0);
    ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(10);
    scheduledExecutorService.scheduleWithFixedDelay(()->{
        //计数加1
        n.incrementAndGet();
        //第四次抛出异常 用于测试抛出异常后不再重复操作
        if (n.get()==4){ int a=0; System.out.println(1/a); }

        //测试执行时间
        try {
            System.out.println("第"+n+"次打印 与启动时相差:"+(new Date().getTime()-millisecondsNow)/1000+"秒");
            //睡眠相当于业务执行时间4秒
            Thread.sleep(4000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 3000 /*延时3秒后首次执行*/, 1000 /*任务间隔1秒*/, TimeUnit.MILLISECONDS);
}
结果:
第1次打印 与启动时相差:3秒
第2次打印 与启动时相差:8秒
第3次打印 与启动时相差:13秒

从上面结果可以看出
1 时间间隔为4秒,时间差等于 任务间隔时间(1秒)+任务处理时间(4秒)
2 不会出现前后两次并发的情景出现
3 第四次抛出异常后不再继续执行任务

类图：

从类图中可以分析出队列关注的是Runnable接口,也即具体执行任务是什么, ScheduledFutureTask实现了多个接口, 实际关心者是持有者例: RunnableFuture<T> ftask = newTaskFor(task, result)

 

 

 

 

 

 

源码工作流程简述：

1 创建一个delayBlockQueue延迟阻塞队列, 队列的功能是take()的时候根据给定条件阻塞是否能取出，offer()的时候根据给定的条件进行排序, 从而得到一个有序队列

2 提交任务时比较核心线程是否全部使用, 若未全部使用则开辟一个新的核心线程, 因为构造函数不能设置delayBlockQueue大小(有的文章说边界), 所以只有核心线程会参与执行任务

3 和ThreadPoolExcutor一样, 核心线程会不断的从队列中取出任务,

判断任务是否为周期任务

1非周期任务则直接执行任务

2若是周期任务则根据周期时间正负决定下一次执行时间 a.若正 执行时间=上一次执行时间+延时 b.若负 执行时间=当前时间+延时 

可以看出如果需要每次执行完都有延时 则周期时间为负数, 若需要当执行时间大于延时时间则需要设置周期时间为正数

数据结构图：

关键源码分析：

 ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit)

ScheduledThreadPoolExecutor extends ThreadPoolExecutor

//构造函数
public ScheduledThreadPoolExecutor(int corePoolSize) {
    //实际上是使用了ThreadPoolExecutor的构造方法 阻塞队列为DelayedWorkQueue(有序的延迟队列)
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

//-----------------------------提交延时任务----------------------------
public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    //构建任务Task
    RunnableScheduledFuture<?> t = decorateTask(command,
    new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));
    //延时执行任务
    delayedExecute(t);
    return t;
}

//执行时间 公式很强简单 任务执行时间=当前时间+延时时间
private long triggerTime(long delay, TimeUnit unit) {
    return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}

//延时执行任务
private void delayedExecute(RunnableScheduledFuture<?> task) {
    //判断线程池状态是否为shutDown 若是则拒绝此次任务
    if (isShutdown())
        //拒绝任务
        reject(task);
    else {
        //将任务加入队列 注意:DelayedWorkQueue入队列需要根据任务执行时间排序
        super.getQueue().add(task);
        //判断是否需要取消此次任务
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
            //确保任务能运行
            ensurePrestart();
    }
}

//判断是否能在此状态下运行
boolean canRunInCurrentRunState(boolean periodic) {
    // 返回是否可以运行 
    // 1若是周期执行根据continueExistingPeriodicTasksAfterShutdown判断
    // 2若非周期函数根据executeExistingDelayedTasksAfterShutdown判断
    return isRunningOrShutdown(periodic ?
                               continueExistingPeriodicTasksAfterShutdown :
                               executeExistingDelayedTasksAfterShutdown);
}
//从队列中移除此次任务
public boolean remove(Runnable task) {
    boolean removed = workQueue.remove(task);
    tryTerminate(); // In case SHUTDOWN and now empty
    return removed;
}

//确保任务能运行
void ensurePrestart() {
    //获取当前worker数量
    int wc = workerCountOf(ctl.get());
    //如果数量小于核心线程数 则开启新的核心工作线程
    if (wc < corePoolSize)
        addWorker(null, true);
    //如果根本没有工作中的线程 则开启一个非核心线程
    else if (wc == 0)
        addWorker(null, false);
}

//-----------------------------执行延时任务----------------------------
//上一节我们知道工作线程向阻塞队列获取任务 然后执行task.run() 来完成任务执行 
//下面直接看ScheduledFutureTask的run方法
public void run() {
    //判断是否为周期任务
    boolean periodic = isPeriodic();
    //判断当前周期任务是否可以在线程池shutDown之后继续执行 根据continueExistingPeriodicTasksAfterShutdown判断
    if (!canRunInCurrentRunState(periodic))
        //删除任务
        cancel(false);
    //如果非周期任务 则与普通任务一样直接执行
    else if (!periodic)
        ScheduledFutureTask.super.run();
    //如果此任务为周期任务 则需要考虑执行完之后重新放入执行队列 保证下一次执行
    else if (ScheduledFutureTask.super.runAndReset()) {
        //设置下一次执行时间
        setNextRunTime();
        //重新放入任务队列 等待工作工作线程获取任务
        reExecutePeriodic(outerTask);
    }
}
//设置下一次远行时间
//上面有讲到过两个周期执行方法的区别:
//scheduleAtFixedRate执行时间若大于周期时间 则执行完此次任务 直接执行下一次任务 不需要延时 
//scheduleWithFixedDelay 不管执行时间是否大于周期时间 都需要在上一次执行完成之后延迟执行
//其原理就在下面
private void setNextRunTime() {
    //scheduleAtFixedRate设置period为正数 scheduleWithFixedDelay设置period为负数
    long p = period;
    if (p > 0)
        time += p;
    else
        //当前时间基础上+延迟时间
        time = triggerTime(-p);
}

 

 思考:

以上源码解析基本解释了计划任务线程池的工作原理, 并且使用了一个阻塞的有序延迟队列来维护任务计划任务的存放, 既然有序肯定需要根据某个值来进行对比,然后排序

已经知道是根据执行时间time(当前时间+延迟时间)来判断排序的,  但是若一样如何解决呢, 线程池另外维护个一个原子参数 AtomicLong sequencer , 线程池内所有新增任务共享,

每次新增任务则sequencer加1, 并把赋值给task的属性long sequenceNumber(任务非共享不需要原子约束),继而每个任务都有一个排序号,当执行时间time值一样时则根据sequenceNumber对比

ScheduledThreadPoolExecutor 与 ThreadPoolExecutor区别

1 ScheduledThreadPoolExecutor 使用的延时阻塞排序队列(类DelayedWorkQueue),根据执行时间排序; 而 ThreadPoolExecutor使用的的是阻塞队列(接口BlockingQueue),定义上BlockingQueue更抽象, 可以支持任何实现BlockingQueue接口的实现类

2 ScheduledThreadPoolExecutor 可以根据任务的属性重新加入队列等待下一次执行; ThreadPoolExecutor是一次性任务

3 ScheduledThreadPoolExecutor 队列不设边界, 也无法设置边界, 任务数非常多可能导致内存不够;  ThreadPoolExecutor可以设置任务队列边界, 任务过多时丢弃新的任务