线程池之ScheduledThreadPoolExecutor

时间  2019-11-08
标签 线程 scheduledthreadpoolexecutor 繁體版
原文   原文链接

1. ScheduledThreadPoolExecutor简介

ScheduledThreadPoolExecutor能够用来在给定延时后执行异步任务或者周期性执行任务，相对于任务调度的Timer来讲，其功能更增强大，Timer只能使用一个后台线程执行任务，而ScheduledThreadPoolExecutor则能够经过构造函数来指定后台线程的个数。ScheduledThreadPoolExecutor类的UML图以下：数组

ScheduledThreadPoolExecutor类的UML图.png

  1. 从UML图能够看出，ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，也就是说ScheduledThreadPoolExecutor拥有execute()和submit()提交异步任务的基础功能，关于ThreadPoolExecutor能够看这篇文章。可是，ScheduledThreadPoolExecutor类实现了ScheduledExecutorService，该接口定义了ScheduledThreadPoolExecutor可以延时执行任务和周期执行任务的功能；
  2. ScheduledThreadPoolExecutor也两个重要的内部类：DelayedWorkQueue和ScheduledFutureTask。能够看出DelayedWorkQueue实现了BlockingQueue接口，也就是一个阻塞队列，ScheduledFutureTask则是继承了FutureTask类，也表示该类用于返回异步任务的结果。这两个关键类，下面会具体详细来看。

1.1 构造方法

ScheduledThreadPoolExecutor有以下几个构造方法：数据结构

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}；

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

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

能够看出因为ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，它的构造方法其实是调用了ThreadPoolExecutor，对ThreadPoolExecutor的介绍能够能够看这篇文章，理解ThreadPoolExecutor构造方法的几个参数的意义后，理解这就很容易了。能够看出，ScheduledThreadPoolExecutor的核心线程池的线程个数为指定的corePoolSize，当核心线程池的线程个数达到corePoolSize后，就会将任务提交给有界阻塞队列DelayedWorkQueue，对DelayedWorkQueue在下面进行详细介绍，线程池容许最大的线程个数为Integer.MAX_VALUE，也就是说理论上这是一个大小无界的线程池。异步

1.2 特有方法

ScheduledThreadPoolExecutor实现了ScheduledExecutorService接口，该接口定义了可延时执行异步任务和可周期执行异步任务的特有功能，相应的方法分别为：ide

//达到给定的延时时间后，执行任务。这里传入的是实现Runnable接口的任务，
//所以经过ScheduledFuture.get()获取结果为null
public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);
//达到给定的延时时间后，执行任务。这里传入的是实现Callable接口的任务，
//所以，返回的是任务的最终计算结果
 public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

//是以上一个任务开始的时间计时，period时间过去后，
//检测上一个任务是否执行完毕，若是上一个任务执行完毕，
//则当前任务当即执行，若是上一个任务没有执行完毕，则须要等上一个任务执行完毕后当即执行
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
//当达到延时时间initialDelay后，任务开始执行。上一个任务执行结束后到下一次
//任务执行，中间延时时间间隔为delay。以这种方式，周期性执行任务。
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);
复制代码

2. 可周期性执行的任务---ScheduledFutureTask

ScheduledThreadPoolExecutor最大的特点是可以周期性执行异步任务，当调用schedule,scheduleAtFixedRate和scheduleWithFixedDelay方法时，其实是将提交的任务转换成的ScheduledFutureTask类，从源码就能够看出。以schedule方法为例：函数

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;
}
复制代码

能够看出，经过decorateTask会将传入的Runnable转换成ScheduledFutureTask类。线程池最大做用是将任务和线程进行解耦，线程主要是任务的执行者，而任务也就是如今所说的ScheduledFutureTask。紧接着，会想到任何线程执行任务，总会调用run()方法。为了保证ScheduledThreadPoolExecutor可以延时执行任务以及可以周期性执行任务，ScheduledFutureTask重写了run方法：源码分析

public void run() {
    boolean periodic = isPeriodic();
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    else if (!periodic)
		//若是不是周期性执行任务，则直接调用run方法
        ScheduledFutureTask.super.run();
		//若是是周期性执行任务的话，须要重设下一次执行任务的时间
    else if (ScheduledFutureTask.super.runAndReset()) {
        setNextRunTime();
        reExecutePeriodic(outerTask);
    }
}
复制代码

从源码能够很明显的看出，在重写的run方法中会先if (!periodic)判断当前任务是不是周期性任务，若是不是的话就直接调用run()方法；不然的话执行setNextRunTime()方法重设下一次任务执行的时间，并经过reExecutePeriodic(outerTask)方法将下一次待执行的任务放置到DelayedWorkQueue中。post

所以，能够得出结论：ScheduledFutureTask最主要的功能是根据当前任务是否具备周期性，对异步任务进行进一步封装。若是不是周期性任务（调用schedule方法）则直接经过run()执行，如果周期性任务，则须要在每一次执行完后，重设下一次执行的时间，而后将下一次任务继续放入到阻塞队列中。idea

3. DelayedWorkQueue

在ScheduledThreadPoolExecutor中还有另外的一个重要的类就是DelayedWorkQueue。为了实现其ScheduledThreadPoolExecutor可以延时执行异步任务以及可以周期执行任务，DelayedWorkQueue进行相应的封装。DelayedWorkQueue是一个基于堆的数据结构，相似于DelayQueue和PriorityQueue。在执行定时任务的时候，每一个任务的执行时间都不一样，因此DelayedWorkQueue的工做就是按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面。spa

为何要使用DelayedWorkQueue呢？线程

定时任务执行时须要取出最近要执行的任务，因此任务在队列中每次出队时必定要是当前队列中执行时间最靠前的，因此天然要使用优先级队列。

DelayedWorkQueue是一个优先级队列，它能够保证每次出队的任务都是当前队列中执行时间最靠前的，因为它是基于堆结构的队列，堆结构在执行插入和删除操做时的最坏时间复杂度是 O(logN)。

DelayedWorkQueue的数据结构

//初始大小
private static final int INITIAL_CAPACITY = 16;
//DelayedWorkQueue是由一个大小为16的数组组成，数组元素为实现RunnableScheduleFuture接口的类
//实际上为ScheduledFutureTask
private RunnableScheduledFuture<?>[] queue =
    new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
private final ReentrantLock lock = new ReentrantLock();
private int size = 0;
复制代码

能够看出DelayedWorkQueue底层是采用数组构成的，关于DelayedWorkQueue能够看这篇博主的文章，很详细。

关于DelayedWorkQueue咱们能够得出这样的结论：DelayedWorkQueue是基于堆的数据结构，按照时间顺序将每一个任务进行排序，将待执行时间越近的任务放在在队列的队头位置，以便于最早进行执行。

4.ScheduledThreadPoolExecutor执行过程

如今咱们对ScheduledThreadPoolExecutor的两个内部类ScheduledFutueTask和DelayedWorkQueue进行了了解，实际上这也是线程池工做流程中最重要的两个关键因素：任务以及阻塞队列。如今咱们来看下ScheduledThreadPoolExecutor提交一个任务后，总体的执行过程。以ScheduledThreadPoolExecutor的schedule方法为例，具体源码为：

public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
	//将提交的任务转换成ScheduledFutureTask
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Void>(command, null,
                                      triggerTime(delay, unit)));
    //延时执行任务ScheduledFutureTask
	delayedExecute(t);
    return t;
}
复制代码

方法很容易理解，为了知足ScheduledThreadPoolExecutor可以延时执行任务和能周期执行任务的特性，会先将实现Runnable接口的类转换成ScheduledFutureTask。而后会调用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
			//保证至少有一个线程启动，即便corePoolSize=0
            ensurePrestart();
    }
}
复制代码

delayedExecute方法的主要逻辑请看注释，能够看出该方法的重要逻辑会是在ensurePrestart()方法中，它的源码为：

void ensurePrestart() {
    int wc = workerCountOf(ctl.get());
    if (wc < corePoolSize)
        addWorker(null, true);
    else if (wc == 0)
        addWorker(null, false);
}
复制代码

能够看出该方法逻辑很简单，关键在于它所调用的addWorker方法，该方法主要功能：新建Worker类，当执行任务时，就会调用被Worker所重写的run方法，进而会继续执行runWorker方法。在runWorker方法中会调用getTask方法从阻塞队列中不断的去获取任务进行执行，直到从阻塞队列中获取的任务为null的话，线程结束终止。addWorker方法是ThreadPoolExecutor类中的方法，对ThreadPoolExecutor的源码分析能够看这篇文章，很详细。

5.总结

  1. ScheduledThreadPoolExecutor继承了ThreadPoolExecutor类，所以，总体上功能一致，线程池主要负责建立线程（Worker类），线程从阻塞队列中不断获取新的异步任务，直到阻塞队列中已经没有了异步任务为止。可是相较于ThreadPoolExecutor来讲，ScheduledThreadPoolExecutor具备延时执行任务和可周期性执行任务的特性，ScheduledThreadPoolExecutor从新设计了任务类ScheduleFutureTask,ScheduleFutureTask重写了run方法使其具备可延时执行和可周期性执行任务的特性。另外，阻塞队列DelayedWorkQueue是可根据优先级排序的队列，采用了堆的底层数据结构，使得与当前时间相比，待执行时间越靠近的任务放置队头，以便线程可以获取到任务进行执行；

  2. 线程池不管是ThreadPoolExecutor仍是ScheduledThreadPoolExecutor，在设计时的三个关键要素是：任务，执行者以及任务结果。它们的设计思想也是彻底将这三个关键要素进行了解耦。

    执行者

    任务的执行机制，彻底交由Worker类，也就是进一步了封装了Thread。向线程池提交任务，不管为ThreadPoolExecutor的execute方法和submit方法，仍是ScheduledThreadPoolExecutor的schedule方法，都是先将任务移入到阻塞队列中，而后经过addWork方法新建了Work类，并经过runWorker方法启动线程，并不断的从阻塞对列中获取异步任务执行交给Worker执行，直至阻塞队列中没法取到任务为止。

    任务

    在ThreadPoolExecutor和ScheduledThreadPoolExecutor中任务是指实现了Runnable接口和Callable接口的实现类。ThreadPoolExecutor中会将任务转换成FutureTask类，而在ScheduledThreadPoolExecutor中为了实现可延时执行任务和周期性执行任务的特性，任务会被转换成ScheduledFutureTask类，该类继承了FutureTask，并重写了run方法。

    任务结果

    在ThreadPoolExecutor中提交任务后，获取任务结果能够经过Future接口的类，在ThreadPoolExecutor中实际上为FutureTask类，而在ScheduledThreadPoolExecutor中则是ScheduledFutureTask类

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程