ThreadPoolExecutor之四：jdk实现的ScheduledThreadPoolExecutor

一、定时任务调度方式常见的有：

1、cron是一个linux下的定时执行工具，一些重要的任务的定时执行可以通过cron来实现，例如每天凌晨1点备份数据等。

2、在JAVA WEB开发中，我们也经常需要用到定时执行任务的功能，JDK提供了Timer类与ScheduledThreadPoolExecutor类实现这个定时功能。

3、quartz

4、分布式任务调度平台XXL-JOB(http://www.cnblogs.com/xuxueli/p/5021979.html)

　　Timer有着不少缺陷，如Timer是单线程模式，调度多个周期性任务时，如果某个任务耗时较久就会影响其它任务的调度；如果某个任务出现异常而没有被catch则可能导致唯一的线程死掉而所有任务都不会再被调度。ScheduledThreadPoolExecutor解决了很多Timer存在的缺陷。

二、ScheduledThreadPoolExecutor使用示例

本章节主要讲述ScheduledThreadPoolExecutor，先通过一个示例了解下基本用法：

package com.dxz.concurrent.schedulethreadpool;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.ScheduledThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolExecutorTest {

    public static void main(String[] args) {

        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);

        BusinessTask task = new BusinessTask();

        // 1秒后开始执行任务，以后每隔2秒执行一次

        executorService.scheduleWithFixedDelay(task, 1000, 2000, TimeUnit.MILLISECONDS);

    }

    private static class BusinessTask implements Runnable {

        @Override

        public void run() {

            System.out.println("任务开始...");

            doBusiness();

            System.out.println("任务结束...");

        }

        private void doBusiness() {

            /*String a = null;

            System.out.println("a:="+a.toString());*/

        }

    }

}

结果(片段)：

...

任务结束...

任务开始...

任务结束...

任务开始...

任务结束...

任务开始...

...

如果将上面的doBusiness()中的注释去掉，

结果：（一直被卡住了）

任务开始...

调度功能不起作用，原因：

创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。如果任务的任一执行遇到异常，就会取消后续执行。否则，只能通过执行程序的取消或终止方法来终止该任务。这样就需要把doBusiness()方法的所有可能异常捕获，才能保证定时任务继续执行。

三、ScheduledThreadPoolExecutor原理

先来看看ScheduledThreadPoolExecutor的实现模型，它通过继承ThreadPoolExecutor来重用线程池的功能，里面做了几件事情：

为线程池设置了一个DelayedWorkQueue，该queue同时具有PriorityQueue（优先级大的元素会放到队首）和DelayQueue（如果队列里第一个元素的getDelay返回值大于0，则take调用会阻塞）的功能
将传入的任务封装成ScheduledFutureTask，这个类有两个特点，实现了java.lang.Comparable和java.util.concurrent.Delayed接口，也就是说里面有两个重要的方法：compareTo和getDelay。ScheduledFutureTask里面存储了该任务距离下次调度还需要的时间（使用的是基于System#nanoTime实现的相对时间，不会因为系统时间改变而改变，如距离下次执行还有10秒，不会因为将系统时间调前6秒而变成4秒后执行）。getDelay方法就是返回当前时间（运行getDelay的这个时刻）距离下次调用之间的时间差；compareTo用于比较两个任务的优先关系，距离下次调度间隔较短的优先级高。那么，当有任务丢进上面说到的DelayedWorkQueue时，因为它有DelayQueue（DelayQueue的内部使用PriorityQueue来实现的）的功能，所以新的任务会与队列中已经存在的任务进行排序，距离下次调度间隔短的任务排在前面，也就是说这个队列并不是先进先出的；另外，在调用DelayedWorkQueue的take方法的时候，如果没有元素，会阻塞，如果有元素而第一个元素的getDelay返回值大于0（前面说过已经排好序了，第一个元素的getDelay不会大于后面元素的getDelay返回值），也会一直阻塞。

ScheduledFutureTask提供了一个run的实现，线程池执行的就是这个run方法。看看run的源码（本文的代码取自hotspot1.5.0_22，jdk后续版本的代码可能已经不一样了，如jdk1.7中使用了自己实现的DelayedWorkQueue，而不再使用PriorityQueue作为存储，不过从外面看它们的行为还是一样的，所以并不影响对ScheduledThreadPoolExecutor调度机制的理解）：

public void run() {

            boolean periodic = isPeriodic();

            if (!canRunInCurrentRunState(periodic))

                cancel(false);

            else if (!periodic)//2

                ScheduledFutureTask.super.run();

            else if (ScheduledFutureTask.super.runAndReset()) {

                setNextRunTime();

                reExecutePeriodic(outerTask);

            }

        }

        public boolean cancel(boolean mayInterruptIfRunning) {

            boolean cancelled = super.cancel(mayInterruptIfRunning);

            if (cancelled && removeOnCancel && heapIndex >= 0)

                remove(this);

            return cancelled;

        }

综上，可以看到，ScheduledThreadPoolExecutor执行周期性任务的模型就是：调度一次任务，计算并设置该任务下次间隔，将任务放回队列中供线程池执行。这里的队列起了很大的作用，且有一些特点：距离下次调度间隔短的任务总是在队首，队首的任务若距离下次调度的间隔时间大于0就无法从该队列的take()方法中拿到任务。

周期性任务的入口（

ScheduledThreadPoolExecutor.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)和

ScheduledThreadPoolExecutor.scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

）与非周期性任务是类似的，它们处理方式不同的地方在于前文说到的ScheduledFutureTask#run()中。

scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,

                                                  long initialDelay,

                                                  long period,

                                                  TimeUnit unit) {

        if (command == null || unit == null)

            throw new NullPointerException();

        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);

        sft.outerTask = t;

        delayedExecute(t);

        return t;

    }

固定频率的调度方法：scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

    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),

                                          unit.toNanos(-delay));

        RunnableScheduledFuture<Void> t = decorateTask(command, sft);

        sft.outerTask = t;

        delayedExecute(t);

        return t;

    }

接下来看看ScheduledThreadPoolExecutor.schedule(Callable callable, long delay, TimeUnit unit)和ScheduledThreadPoolExecutor.schedule(Runnable command, long delay, TimeUnit unit)这两个非周期性任务的实现方式，先看看它们的源码：

     public <V> ScheduledFuture<V> schedule(Callable<V> callable,

                                           long delay,

                                           TimeUnit unit) {

        if (callable == null || unit == null)

            throw new NullPointerException();

        RunnableScheduledFuture<V> t = decorateTask(callable,

            new ScheduledFutureTask<V>(callable,

                                       triggerTime(delay, unit)));

        delayedExecute(t);

        return t;

    }

    public ScheduledFuture<?> schedule(Runnable command,

                                       long delay,

                                       TimeUnit unit) {

        if (command == null || unit == null)

            throw new NullPointerException();

        RunnableScheduledFuture<?> t = decorateTask(command,

            new ScheduledFutureTask<Void>(command, null,

                                          triggerTime(delay, unit)));

        delayedExecute(t);

        return t;

    }

delayedExecute()延迟或周期任务的主要执行方法:

 //延迟或周期任务的主要执行方法。如果线程池关闭，则拒绝任务。否则，将任务添加到队列中，并在必要时启动一个线程来运行它。（我们不能启动线程运行任务因为任务（可能）不应该运行。）如果线程池关闭，任务被添加，如果需要取消，由状态和运行后关闭参数删除它。

 private void delayedExecute(RunnableScheduledFuture<?> task) {

        if (isShutdown())

            reject(task);

        else {

            super.getQueue().add(task);

            if (isShutdown() &&

                !canRunInCurrentRunState(task.isPeriodic()) &&

                remove(task))

                task.cancel(false);

            else

                ensurePrestart();

        }

    }

实现方式也很简单，在创建ScheduledThreadPoolExecutor内部任务（即ScheduledFutureTask）的时候就将调度间隔计算并设置好，如果当前线程数小于设置的核心线程数，就启动一个线程（可能是线程池刚启动里面还没有线程，也可能是里面的线程执行任务时挂掉了。如果线程池中的线程挂掉了而又没有调用这些schedule方法谁去补充挂掉的线程？不用担心，线程池自己会处理的）去监听队列里的任务，然后将任务放到队列里，在任务执行间隔不大于0的时候，线程就可以拿到这个任务并执行。

巴特西

ThreadPoolExecutor之四：jdk实现的ScheduledThreadPoolExecutor

一、定时任务调度方式常见的有：

二、ScheduledThreadPoolExecutor使用示例

三、ScheduledThreadPoolExecutor原理

最新文章

热门文章