标签：epo   his   reset   news   remove   周期性任务   支持   val   必须   

【原创】Java并发编程系列35 | ScheduledThreadPoolExecutor定时器

收录于话题
#进阶架构师 | 并发编程专题 12 #并发 2 #java 2
★★★建议星标我们★★★
公众号改版后文章乱序推荐，希望你可以点击上方“Java进阶架构师”，点击右上角，将我们设为★“星标”！这样才不会错过每日进阶架构文章呀。

2020年Java原创面试题库连载中
【000期】Java最全面试题库思维导图
【001期】JavaSE面试题（一）：面向对象
【002期】JavaSE面试题（二）：基本数据类型与访问修饰符
【003期】JavaSE面试题（三）：JavaSE语法（1）
【004期】JavaSE面试题（四）：JavaSE语法（3）
【005期】JavaSE面试题（五）：String类
【006期】JavaSE面试题（六）：泛型
【007期】JavaSE面试题（七）：异常
【008期】JavaSE面试题（八）：集合之List
【009期】JavaSE面试题（九）：集合之Set
【010期】JavaSE面试题（十）：集合之Map
【011期】JavaSE面试题（十一）：多线程（1）
【012期】JavaSE面试题（十二）：多线程（2）
【013期】JavaSE面试题（十三）：多线程（3）
【014期】JavaSE面试题（十四）：基本IO流
【015期】JavaSE面试题（十五）：网络IO流
【016期】JavaSE面试题（十六）：反射
【017期】JavaSE面试题（十七）：JVM之内存模型
【018期】JavaSE面试题（十八）：JVM之垃圾回收
【020期】JavaSE系列面试题汇总（共18篇）
【019期】JavaWeb面试题（一）：JDBC
【021期】JavaWeb面试题（二）：HTTP协议
【022期】JavaWeb面试题（三）：Cookie和Session
【023期】JavaWeb面试题（四）：JSP
【024期】JavaWeb面试题（五）：Filter和Listener
【025期】Java工具面试题（一）：版本控制工具
【026期】Java工具面试题（二）：项目管理工具
【027期】Java设计模式面试题
【028期】JavaWeb系列面试题汇总（共10篇）
【029期】JavaEE面试题（一）Web应用服务器
【030期】JavaEE面试题（二）SpringMVC
【031期】JavaEE面试题（三）Spring（1）
【032期】JavaEE面试题（四）Spring（2）
【033期】JaveEE面试题（五）MyBatis
【034期】JavaEE面试题（六）Hibernate
【035期】JavaEE面试题（七）SpringBoot（1）
更多内容，点击上面蓝字查看

上一篇讲解了线程池的原理，这篇就在线程池基础上介绍基于线程池实现的定时器ScheduledThreadPoolExecutor：
ScheduledThreadPoolExecutor的用法
ScheduledThreadPoolExecutor源码分析
ScheduledThreadPoolExecutor执行过程分析

1. 介绍

---

ScheduledThreadPoolExecutor 可以用来在给定延时后执行异步任务或者周期性执行任务，也就是我们说的定时器。ScheduledThreadPoolExecutor基于线程池，通过多线程实现延时和周期执行。

1.1 用法Demo

如下代码使用ScheduledThreadPoolExecutor实现：10ms后打印第一次，之后每隔30ms打印一次。

public class TimerDemo {
    public static void main(String[] args) {
        // 1. 创建线程池定时器
        ScheduledExecutorService timer = Executors.newScheduledThreadPool(3);
        // 2. 提交定时任务：10ms后打印第一次，之后每隔30ms打印一次
        timer.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println(1);
            }
        }, 10, 30, TimeUnit.MILLISECONDS);
    }
}

1.2 四种定时器用法

第一种schedule(Runnable command, long delay, TimeUnit unit)；达到给定的延时时间后，执行任务，Runnable不能返回结果；
第二种schedule(Callablecallable, long delay, TimeUnit unit); 达到给定的延时时间后，执行任务，Callable可以返回结果；
第三种scheduleAtFixedRate(); 固定周期执行任务，每次执行的开始时间之间的间隔是固定的，最开始就能够确定之后每次执行的时间；
第四种scheduleWithFixedDelay(); 固定延时周期执行任务，上一次执行结束到下一次执行开始的间隔时间是固定的，由于每次执行任务花费时间不一定相同，所以只有在上次执行结束之后才能确定下次执行开始的时间。

    /**
     * 达到给定的延时时间后，执行任务
     * @param command Runnable接口的任务，ScheduledFuture.get()获取结果为null
     */
    public ScheduledFuture> schedule(Runnable command, long delay, TimeUnit unit);

    /**
     * 到给定的延时时间后，执行任务。
     * @param callable 实现Callable接口的任务，ScheduledFuture.get()可获取任务结果
     */
    public  ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit);

    /**
     * 固定周期执行任务
     * @param initialDelay 第一次执行的延迟时间
     * @param period 周期
     */
    public ScheduledFuture> scheduleAtFixedRate(Runnable command,
                    long initialDelay, long period, TimeUnit unit);

    /**
     * 固定延时周期执行任务
     * @param initialDelay 第一次执行的延迟时间
     * @param delay 上一次执行结束到下一次执行开始的间隔时间
     */
    public ScheduledFuture> scheduleWithFixedDelay(Runnable command,
                    long initialDelay, long delay, TimeUnit unit);

2. 类结构

---

2.1 继承结构

ScheduledThreadPoolExecutor 继承了ThreadPoolExecutor，是一种特殊的线程池，拥有 execute()和 submit()提交异步任务功能。
ScheduledThreadPoolExecutor 类实现了ScheduledExecutorService，该接口定义了延时执行任务和周期执行任务的功能；
ScheduledThreadPoolExecutor 有两个重要的内部类：DelayedWorkQueue和ScheduledFutureTask。DelayedWorkQueue 实现了 BlockingQueue 接口，是一个阻塞队列；ScheduledFutureTask 继承了 FutureTask 类，是一个可以返回异步任务的结果的Runnable。

2.2 构造方法

通过构造方法可以看到，创建ScheduledThreadPoolExecutor其实就是创建一个线程池，corePoolSize可以指定，maximumPoolSize为Integer.MAX_VALUE，任务队列为DelayedWorkQueue。

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

2.3 ScheduledFutureTask

ScheduledThreadPoolExecutor 提交的任务时，将任务封装成 ScheduledFutureTask，当执行任务时通过ScheduledFutureTask的run()方法调用任务的run()方法。
ScheduledFutureTask的compareTo()方法用于延迟队列排序，按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面。

private class ScheduledFutureTask 
        extends FutureTask implements RunnableScheduledFuture {
    private long time;// 下次执行时间
    private final long period;// 周期

    /**
     * 比较方法，到下次执行时间短的任务优先
     */
    public int compareTo(Delayed other) {
        if (other == this) // compare zero if same object
            return 0;
        if (other instanceof ScheduledFutureTask) {
            ScheduledFutureTask> x = (ScheduledFutureTask>)other;
            long diff = time - x.time;
            if (diff  0)
                return 1;
            else if (sequenceNumber  0) ? 1 : 0;
    }

    /**
     * ScheduledThreadPoolExecutor提交的任务被封装成ScheduledFutureTask，所以任务执行要通过这个run()方法
     */
    public void run() {
        boolean periodic = isPeriodic();// 是否是周期执行
        // 当前线程池运行状态下如果不可以执行任务，取消该任务
        if (!canRunInCurrentRunState(periodic))
            cancel(false);
        // 如果不是周期性任务，直接调用FutureTask中的run方法执行
        else if (!periodic)
            ScheduledFutureTask.super.run();
        // 如果是周期性任务，调用FutureTask中的runAndReset方法执行
        else if (ScheduledFutureTask.super.runAndReset()) {
            setNextRunTime();// 计算下次执行该任务的时间
            reExecutePeriodic(outerTask);// 将新任务再次放入线程池等待被执行任务
        }
    }
}

2.4 DelayedWorkQueue

DelayedWorkQueue 是一个基于小顶堆的数据结构，类似于 DelayQueue 和 PriorityQueue。DelayedWorkQueue 按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面。之前已经详细介绍过 DelayQueue 和 PriorityQueue了，这里就不在重复了。
DelayedWorkQueue是存储ScheduledFutureTask阻塞队列。
插入元素时，会根据延期时间对元素排序，队头的元素是最先到期的；
取出元素时，只有在队头元素到期时才能够从队列中取元素。如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。

static class DelayedWorkQueue extends AbstractQueue
    implements BlockingQueue {
    // 队列初始容量
    private static final int INITIAL_CAPACITY = 16;
    // 根据初始容量创建RunnableScheduledFuture类型的数组
    private RunnableScheduledFuture>[] queue =
            new RunnableScheduledFuture>[INITIAL_CAPACITY];
    private final ReentrantLock lock = new ReentrantLock();
    private int size = 0;
    //leader线程
    private Thread leader = null;
    //当较新的任务在队列的头部可用时，或者新线程可能需要成为leader，则通过该条件发出信号
    private final Condition available = lock.newCondition();

    /**
     * 根据延期时间对元素排序，队头的元素是最先到期的
     */
    public boolean offer(Runnable x) {}
    /**
     * 只有在队头元素到期时才能够从队列中取元素。
     * 如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。
     */
    public RunnableScheduledFuture> take() throws InterruptedException {}

    // ......
}

3. 执行过程

---

以上文的小示例为例，通过源码来分析ScheduledThreadPoolExecutor的执行过程。
示例代码：

public class TimerDemo {
    public static void main(String[] args) {
        // 1. 创建线程池定时器
        ScheduledExecutorService timer = Executors.newScheduledThreadPool(3);
        // 2. 提交定时任务：10ms后打印第一次，之后每隔30ms打印一次
        timer.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println(1);
            }
        }, 10, 30, TimeUnit.MILLISECONDS);
    }
}

3.1 创建线程池定时器

创建线程池定时器就是创建一个线程池：

/**
 * Executors.newScheduledThreadPool(3);
 */
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

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

public ThreadPoolExecutor(int corePoolSize,
        int maximumPoolSize,
        long keepAliveTime,
        TimeUnit unit,
        BlockingQueue workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}

3.2 添加任务

Runable任务封装成ScheduledFutureTask；
任务加入延时队列，同时在队列中按照执行的时间顺序排序，最先执行的任务在队头；
确保线程池中有活动线程，如果没有就启动一个。

public ScheduledFuture> scheduleAtFixedRate(Runnable command,
  long initialDelay, long period, TimeUnit unit) {
 // 一些检查
 if (command == null || unit == null)
  throw new NullPointerException();
 if (period  sft = new ScheduledFutureTask(command,
   null, triggerTime(initialDelay, unit), unit.toNanos(period));
 RunnableScheduledFuture t = decorateTask(command, sft);
 sft.outerTask = t;// 设置下次执行的任务
 // 将ScheduledFutureTask类型的任务放入线程池延时执行，下文详细介绍
 delayedExecute(t);
 return t;
}

/**
 * 将ScheduledFutureTask类型的任务放入线程池延时执行
 */
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();// 确保线程池中有线程执行
 }
}

/**
 * 确保线程池中有线程执行
 * 只有两种情况会启动线程：
 * 1. 当前线程数小于corePoolSize，以corePoolSize为界限启动一个线程
 * 2. 线程池参数corePoolSize=0且此时线程池中没有线程，以maximumPoolSize为界限启动一个线程
 */
void ensurePrestart() {
 int wc = workerCountOf(ctl.get());
 // 线程数小于corePoolSize，以corePoolSize为界限启动一个线程
 if (wc 

3.3 执行任务

线程池中的活动线程会循环到任务队列中取任务，当队头任务还没到期时，线程阻塞至队头任务到期时间，然后再取任务；
取出任务后执行，因为任务是ScheduledFutureTask类型（添加任务时封装的），执行ScheduledFutureTask.run()；
ScheduledFutureTask.run()执行当前任务，设置下次执行时间并将任务放入线程池；
线程池中的活动线程会循环到任务队列中取任务，...循环...

/**
 * ScheduledThreadPoolExecutor提交的任务被封装成ScheduledFutureTask，所以任务执行要通过这个run()方法
 */
public void run() {
 boolean periodic = isPeriodic();// 是否是周期执行
 // 当前线程池运行状态下如果不可以执行任务，取消该任务
 if (!canRunInCurrentRunState(periodic))
  cancel(false);
 // 如果不是周期性任务，直接调用FutureTask中的run方法执行
 else if (!periodic)
  ScheduledFutureTask.super.run();
 // 如果是周期性任务，调用FutureTask中的runAndReset方法执行
 else if (ScheduledFutureTask.super.runAndReset()) {
  setNextRunTime();// 计算下次执行该任务的时间
  reExecutePeriodic(outerTask);// 将新任务再次放入线程池等待被执行任务
 }
}

/**
 * 计算下次执行该任务的时间
 */
private void setNextRunTime() {
 long p = period;
 if (p > 0)
  time += p;
 else
  time = triggerTime(-p);
}

/**
 * 将新任务再次放入线程池等待被执行任务
 */
void reExecutePeriodic(RunnableScheduledFuture> task) {
 if (canRunInCurrentRunState(true)) {
  super.getQueue().add(task);// 加入延时队列
  // 删除不符合条件任务
  if (!canRunInCurrentRunState(true) && remove(task))
   task.cancel(false);
  else
   ensurePrestart();// 确保线程池中有线程执行
 }
}

3.3 执行过程总结

Runable任务封装成ScheduledFutureTask；
任务加入延时队列，同时在队列中按照执行的时间顺序排序，最先执行的任务在队头；
确保线程池中有活动线程，如果没有就启动一个；
线程池中的活动线程会循环到任务队列中取任务，当队头任务还没到期时，线程阻塞至队头任务到期时间，然后再取任务；
取出任务后执行，因为任务是ScheduledFutureTask类型（添加任务时封装的），执行ScheduledFutureTask.run()；
ScheduledFutureTask.run()执行当前任务，设置下次执行时间并将任务放入线程池；
线程池中的活动线程会循环到任务队列中取任务，...循环...

4. scheduleAtFixedRate() VS scheduleWithFixedDelay()

---

• scheduleAtFixedRate(); 固定周期执行任务，每次执行的开始时间之间的间隔是固定的，最开始就能够确定之后每次执行的时间；
• scheduleWithFixedDelay(); 固定延时周期执行任务，上一次执行结束到下一次执行开始的间隔时间是固定的，由于每次执行任务花费时间不一定相同，所以只有在上次执行结束之后才能确定下次执行开始的时间。
  从源码角度理解scheduleAtFixedRate()和scheduleWithFixedDelay()的不同，由两个细节决定：
  细节一：构造ScheduledFutureTask时，scheduleAtFixedRate传入period(>0)，而scheduleWithFixedDelay传入-delay(

public ScheduledFuture> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period  sft =
            new ScheduledFutureTask(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));// z这里是period
        RunnableScheduledFuture t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

  public ScheduledFuture> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay  sft =
            new ScheduledFutureTask(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));// 这里是-delay
        RunnableScheduledFuture t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

细节二：执行完一次后设置下次执行时间时，

• p>0，scheduleAtFixedRate，下次执行开始时间=上次开始执行时间+周期
• period

    /**
     * 计算下次执行该任务的时间
     */
    private void setNextRunTime() {
        long p = period;
        // p>0，scheduleAtFixedRate，下次执行开始时间=上次开始执行时间+周期
        if (p > 0)
            time += p;
        // period> 1)) ? delay : overflowFree(delay));
    }

5. 总结

---

1.ScheduledThreadPoolExecutor 可以用来在给定延时后执行异步任务或者周期性执行任务，也就是我们说的定时器。ScheduledThreadPoolExecutor基于线程池，通过多线程实现延时和周期执行。
2.ScheduledThreadPoolExecutor的四种用法：

• schedule(Runnable command, long delay, TimeUnit unit)；达到给定的延时时间后，执行任务，Runnable不能返回结果；
• schedule(Callablecallable, long delay, TimeUnit unit); 达到给定的延时时间后，执行任务，Callable可以返回结果；
• 第三种scheduleAtFixedRate(); 固定周期执行任务，每次执行的开始时间之间的间隔是固定的，最开始就能够确定之后每次执行的时间；
• 第四种scheduleWithFixedDelay(); 固定延时周期执行任务，上一次执行结束到下一次执行开始的间隔时间是固定的，由于每次执行任务花费时间不一定相同，所以只有在上次执行结束之后才能确定下次执行开始的时间。
  执行过程：
  Runable任务封装成ScheduledFutureTask；
  任务加入延时队列，同时在队列中按照执行的时间顺序排序，最先执行的任务在队头；
  确保线程池中有活动线程，如果没有就启动一个；
  线程池中的活动线程会循环到任务队列中取任务，当队头任务还没到期时，线程阻塞至队头任务到期时间，然后再取任务；
  取出任务后执行，因为任务是ScheduledFutureTask类型（添加任务时封装的），执行ScheduledFutureTask.run()；
  ScheduledFutureTask.run()执行当前任务，设置下次执行时间并将任务放入线程池；
  线程池中的活动线程会循环到任务队列中取任务，...循环...

并发系列文章汇总

---

【原创】01|开篇获奖感言
【原创】02|并发编程三大核心问题
【原创】03|重排序-可见性和有序性问题根源
【原创】04|Java 内存模型详解
【原创】05|深入理解 volatile
【原创】06|你不知道的 final
【原创】07|synchronized 原理
【原创】08|synchronized 锁优化
【原创】09|基础干货
【原创】10|线程状态
【原创】11|线程调度
【原创】12|揭秘 CAS
【原创】13|LockSupport
【原创】14|AQS 源码分析
【原创】15|重入锁 ReentrantLock
【原创】16|公平锁与非公平锁
【原创】17|读写锁八讲（上）
【原创】18|读写锁八讲（下）
【原创】19|JDK8新增锁StampedLock
【原创】20|StampedLock源码解析
【原创】21|Condition-Lock的等待通知
【原创】22|倒计时器CountDownLatch
【原创】22|倒计时器CountDownLatch
【原创】23|循环屏障CyclicBarrier
【原创】24|信号量Semaphore
【原创】25|交换器Exchangere
【原创】26|ConcurrentHashMap（上）
【原创】27|ConcurrentHashMap（下）
【原创】28|Copy-On-Write容器
【原创】29|ConcurrentLinkedQueue
【原创】30 | ThreadLocal
【原创】31 | 阻塞队列（上）
【原创】32 | 阻塞队列（下）
【原创】33 | 深入理解线程池（上）
【原创】34 | 深入理解线程池（下）

之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。

获取方式：点“在看”，V信关注上述Java最全面试题库号并回复 【面试】即可领取，更多精彩陆续奉上。

看到这里，证明有所收获
必须点个在看支持呀，喵

【原创】Java并发编程系列35 | ScheduledThreadPoolExecutor定时器

标签：epo   his   reset   news   remove   周期性任务   支持   val   必须   

原文地址：https://blog.51cto.com/15009303/2552571