码迷,mamicode.com
首页 > 编程语言 > 详细

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

时间:2020-11-24 12:58:11      阅读:16      评论:0      收藏:0      [点我收藏+]

标签: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 <V> ScheduledFuture<V> schedule(Callable<V> 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<V> 
        extends FutureTask<V> implements RunnableScheduledFuture<V> {
    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 (diff > 0)
                return 1;
            else if (sequenceNumber < x.sequenceNumber)
                return -1;
            else
                return 1;
        }
        long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
        return (diff < 0) ? -1 : (diff > 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<Runnable>
    implements BlockingQueue<Runnable> {
    // 队列初始容量
    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<Runnable> 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 <= 0)
  throw new IllegalArgumentException();

 // 将command封装成ScheduledFutureTask,包括下次执行时间和周期
 ScheduledFutureTask<Void> sft = new ScheduledFutureTask<Void>(command,
   null, triggerTime(initialDelay, unit), unit.toNanos(period));
 RunnableScheduledFuture<Void> 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 < corePoolSize)
  addWorker(null, true);
 /*
  * 线程数大于等于corePoolSize且线程数==0,其实就是线程池参数corePoolSize=0且此时线程池中没有线程
  * 以maximumPoolSize为界限启动一个线程
  */
 else if (wc == 0)
  addWorker(null, false);
}

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(<0)。
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));// z这里是period
        RunnableScheduledFuture<Void> 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 <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));// 这里是-delay
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

细节二:执行完一次后设置下次执行时间时,

  • p>0,scheduleAtFixedRate,下次执行开始时间=上次开始执行时间+周期
  • period<=0,scheduleWithFixedDelay,下次执行开始时间=上次结束执行时间(now)+周期

    /**
     * 计算下次执行该任务的时间
     */
    private void setNextRunTime() {
        long p = period;
        // p>0,scheduleAtFixedRate,下次执行开始时间=上次开始执行时间+周期
        if (p > 0)
            time += p;
        // period<=0,scheduleWithFixedDelay,下次执行开始时间=上次结束执行时间+周期
        else
            time = triggerTime(-p);
    }

    long triggerTime(long delay) {
        return now() +
                ((delay < (Long.MAX_VALUE >> 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

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!