Java计划线程池ScheduledThreadPoolExecutor运行流程和源码分析

1. 计划线程池ScheduledThreadPoolExecutor简介

ScheduledThreadPoolExecutor继承自线程池ThreadPoolExecutor,并在其基础上增加了按时间调度执行任务的功能,如果对ThreadPoolExecutor还不是很熟悉,可以阅读一下这篇文章:
Java线程池ThreadPoolExecutor运行机制和源码解析。

首先来翻译一下这个类的前面的单词scheduled: 预先安排的,按时刻表的;定期的。那么ScheduledThreadPoolExecutor可翻译为计划线程池、定时线程池。

大多数时候,线程池已经使用了池化技术很好的满足了线程的重复使用需要,为什么还要另外搞一个ScheduledThreadPoolExecutor呢?

普通线程池无法满足以下痛点:

  1. 无法设置任务在指定时间点执行
  2. 无法周期性的执行任务。想要完成这点需要在自己实现,在任务里加一个while循环。

可能有人会问java里已经有了Timer类,能实现定时运行任务。但Timer类有个缺点就是内部只有一个线程,如果有多个任务,在同一时间只有一个任务在运行,无法做到同时运行,虽然可以使用多个Timer解决,但比较难以进行统一的管理。

2. 计划线程池ScheduledThreadPoolExecutor运行机制

ScheduledThreadPoolExecutor在调用提交任务的方法后(比如scheduleAtFixedRatescheduleWithFixedDelay)会将任务放入一个按执行时间升序排行的缓冲队列,工作线程会从任务缓冲队列取任务然后将它执行,在任务完成时会计算下次执行时间然后放回任务缓冲队列(这点是和ThreadPoolExecutor的区别),从而实现周期性执行。

在这里插入图片描述

3. ScheduledThreadPoolExecutor类图

ScheduledThreadPoolExecutor继承自线程池ThreadPoolExecutor,是从ThreadPoolExecutor扩展而来。

在这里插入图片描述

4. ScheduledThreadPoolExecutor构造函数

public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory,RejectedExecutionHandler handler) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue(), threadFactory, handler);
}                                     
参数类型名称备注
corePoolSizeint核心线程数除非设置了allowCoreThreadTimeOut,不然即使空闲也不会回收
threadFactoryThreadFactory线程工厂可设置线程属性
handlerRejectedExecutionHandler拒绝策略当线程容量溢出时执行的策略

ScheduledThreadPoolExecutor的构造函数是调用父类ThreadPoolExecutor的构造函数,区别就是ThreadPoolExecutor可以自行根据功能需要使用不同类型的缓冲队列,而ScheduledThreadPoolExecutor的缓冲队列只能是DelayedWorkQueue

5. 提交计划任务

我们知道,ThreadPoolExecutor提交任务的唯一入口就是execute()方法,任务在execute()方法里进行分配和调度。与之不同的是ScheduledThreadPoolExecutor根据对任务处理需求不同,提交任务有几个方法。


/*** 安排一个任务在延迟一段时间后执行** @param command 任务* @param delay 要延迟多少时间单位执行(从当前时间开始计时)* @param unit delay参数的时间单位*/
public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);/*** 和上面方法的区别是入参任务一个是Runnable,一个Callable*/                            
public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit);/*** 创建并执行一个周期性任务,该任务在指定的延迟时间后首次执行,* 然后以给定的周期重复执行;即任务将在 initialDelay 后开始执行,* 然后在 initialDelay+period,再然后是 initialDelay + 2 * period,以此类推。* 如果任务的任何执行遇到异常,则将禁止后续的执行。* 否则,任务只能通过取消或执行器的终止来终止。* 如果任务的任何执行时间超过其周期,则后续执行可能会延迟开始,但不会并发执行。** @param command 要执行的任务* @param initialDelay 首次执行前的延迟时间* @param period 多次执行的时间周期间隔* @param unit 初始延迟和周期的时间单位*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);/*** 创建并执行一个周期性任务,该任务在指定的延迟时间后首次执行,* 然后在本次执行的终止与下一次执行的开始之间以给定的延迟重复执行。* 如果任务的任何执行遇到异常,则将禁止后续的执行。* 否则,任务只能通过取消或执行器的终止来终止。** @param command 要执行的任务* @param initialDelay 首次执行的延迟时间* @param delay 一个执行的终止与下一个执行的开始之间的延迟* @param unit initialDelay 和 delay 参数的时间单位*/                                                  
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);

提交计划的方法汇总至表格如下:

方法参数参数备注效果
scheduleRunnable command要执行的任务安排一个任务在延迟一段时间后执行
long delay要延迟多少时间单位执行(从当前时间开始计时)
TimeUnit unitdelay参数的时间单位
scheduleCallable<V> callable要执行的callable任务和上面方法的区别是入参任务一个是Runnable,一个Callable
long delay要延迟多少时间单位执行(从当前时间开始计时)
TimeUnit unitdelay参数的时间单位
scheduleAtFixedRateRunnable command要执行的任务创建并执行一个周期性任务,该任务在指定的延迟时间后首次执行;然后以给定的周期重复执行(有可能会连续不间隔的执行)
long initialDelay首次执行前的延迟时间
long period多次执行的时间周期间隔
TimeUnit unit初始延迟和周期的时间单位
scheduleWithFixedDelayRunnable command要执行的任务创建并执行一个周期性任务,该任务在指定的延迟时间后首次执行;然后在本次执行的终止与下一次执行的开始之间 固定以给定的延迟 重复执行。
long initialDelay首次执行前的延迟时间
long period多次执行的时间周期间隔
TimeUnit unit初始延迟和周期的时间单位
scheduleAtFixedRate和scheduleWithFixedDelay的区别
  • scheduleAtFixedRate

以本次执行的开始时间beginTime为基准,下一次的执行的时间为beginTime + period。如果执行任务的时间超过了period,会导致任务完成后无缝再执行一次。

  • scheduleWithFixedDelay

以本次执行的结束时间endTime为基准,下一次的执行的时间为endTime + delay。这个方法无论该次执行使用多少时间,都会在任务执行完成后会等待delay的时间再执行。

6. 阻塞队列DelayedWorkQueue

DelayedWorkQueue提供了一种方便的方式来管理延迟执行的任务,并且能够在高并发情况下安全地进行操作,是实现定时任务调度功能的重要组成部分,以下是DelayedWorkQueue的一些特性:

  1. 延迟执行任务:DelayedWorkQueue用于存储需要延迟执行的任务,这些任务会在指定的延迟时间之后被执行。这使得程序可以安排任务在将来的某个时间点执行,例如定时任务。
  2. 基于优先级队列:DelayedWorkQueue通常是基于优先级队列(PriorityQueue)实现的,其中任务根据其延迟时间被排序。因此,队列中的任务按照延迟的时间进行排列,最先到期的任务将被最先执行。
  3. 数据结构:DelayedWorkQueue通常是一个有序队列,内部使用了二叉堆数据结构来实现。这样可以快速找到下一个要执行的任务,并且在添加新任务时保持队列的有序性。
  4. 并发性:DelayedWorkQueue是线程安全的,内部有一个可重入锁ReentrantLock,通过锁来保证并发安全,它需要被多个线程同时访问,以便添加新任务或者移除已经过期的任务。这涉及到对队列的操作的同步或者使用线程安全的数据结构。
  5. 使用场景:DelayedWorkQueue适用于需要按照一定延迟顺序执行任务的场景,比如定时任务、周期性任务等。它可以作为ScheduledThreadPoolExecutor的内部组件使用,ScheduledThreadPoolExecutor通过DelayedWorkQueue实现了定时任务的调度功能。

DelayedWorkQueue从数据结构来说是一个用数组实现的二叉堆,排序方式为从小到大,每个父节点都比子节点要小的堆也称为最小堆, 堆顶元素是最小的,因此堆顶存储的是执行时间缀最小的任务。

堆(Heap)是一种特殊的树形数据结构,通常实现为数组,其具有以下性质:

  1. 堆序性质:在堆中,对于任意节点 i,如果 j 是 i 的子节点,那么堆中的节点 i 的值必须满足某种特定的顺序关系(比如最小堆中,父节点的值小于或等于子节点的值;最大堆中,父节点的值大于或等于子节点的值)。

  2. 完全二叉树性质:堆通常被实现为完全二叉树,即除了最后一层外,每一层都被完全填满,且最后一层的节点都尽可能地靠左排列。这样的特性使得堆可以利用数组来表示,而不需要使用额外的指针。

由于DelayedWorkQueue是最小堆结构,下面文章着重讲解的是最小堆

最小堆介绍

堆可以分为两种常见的类型:

最小堆(Min Heap):在最小堆中,父节点的值小于或等于其子节点的值,因此堆顶元素是最小的。

最大堆(Max Heap):在最大堆中,父节点的值大于或等于其子节点的值,因此堆顶元素是最大的。

最小堆通常可以用数组来表示,下面是一个列子

最小堆二叉树

在这里插入图片描述

最小堆数组

在这里插入图片描述

对于最小堆数组,有以下的下标对应关系:

节点效果
当前节点arr[i]
父节点arr[(i-1)/2]
左子节点arr[(2*i) + 1]
右子节点arr[(2*i )+ 2]
DelayedWorkQueue的定义
static class DelayedWorkQueue extends AbstractQueue<Runnable>implements BlockingQueue<Runnable> {private static final int INITIAL_CAPACITY = 16;private RunnableScheduledFuture<?>[] queue =new RunnableScheduledFuture<?>[INITIAL_CAPACITY];private final ReentrantLock lock = new ReentrantLock();private int size = 0;    private Thread leader = null;private final Condition available = lock.newCondition();}

类图

在这里插入图片描述

DelayedWorkQueue实现了阻塞队列BlockingQueue,通过可重入锁ReentrantLock保证多线程并发安全,它的最小堆通过数组实现,这个数组初始容量为16,每次扩容增加一半。

核心方法:

  • offer(Runnable x) 向队列添加元素
  • take() 从队列取任务
  • poll(long timeout, TimeUnit unit) 从队列取元素,若在指定时间内未能取到则返回null

offer 向队列增加元素

offer方法其实就是向最小堆添加元素的过程,将元素从最小堆最底部开始上浮,直到比它的parent节点的值要大。上浮完毕后如果该元素在堆顶,则该元素为最先要执行的任务,唤醒等待的一个线程。

offer方法的流程图

在这里插入图片描述

offer方法的代码如下:

public boolean offer(Runnable x) {if (x == null)throw new NullPointerException();RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;final ReentrantLock lock = this.lock;lock.lock();try {int i = size;if (i >= queue.length)grow();     //队列已满的时候进行扩容size = i + 1;if (i == 0) {   //队列为空的时候queue[0] = e;setIndex(e, 0);} else {siftUp(i, e); //元素从底部上浮}if (queue[0] == e) {leader = null;available.signal();   //唤醒一个等待的线程}} finally {lock.unlock();}return true;
}
siftUp 堆排序中的上浮操作

在offer的过程中因为堆添加了新的元素,假设它为e,为了维持最小堆的数据结构,需要将元e素调整到它数值的排序位置上,步骤如下:

  1. 首先将元素e放到堆数据的尾部
  2. 将元素e与父节点比较,若小于父节点则与其交换
  3. 重复步骤2直至元素e大于父子点或元素e已到堆顶
    /*** 堆排序中的上浮操作* 仅当持有锁的时候才会调用.*/private void siftUp(int k, RunnableScheduledFuture<?> key) {//在二叉小堆中,把元素从底向上比较排序while (k > 0) {int parent = (k - 1) >>> 1;  //在二叉堆中元素的父节点位置RunnableScheduledFuture<?> e = queue[parent];if (key.compareTo(e) >= 0)/*** 上浮直到找到比它的parent节点的值要大*/break;queue[k] = e;setIndex(e, k);k = parent;}queue[k] = key;setIndex(key, k);}

take 从队列取元素

take()方法是从堆顶取最小的任务,

  1. 如果队列为空,等待元素入列
  2. 取第一个堆顶最小的元素,计算该元素是否已经到执行时间
  3. 如果已到执行时间,返回元素
  4. 如果未到执行时间,

take()方法流程图:

在这里插入图片描述

源码

/*** 取堆顶元素*/
public RunnableScheduledFuture<?> take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {RunnableScheduledFuture<?> first = queue[0];if (first == null)available.await();else {long delay = first.getDelay(NANOSECONDS);if (delay <= 0)return finishPoll(first);first = null; // don't retain ref while waitingif (leader != null)available.await();else {Thread thisThread = Thread.currentThread();leader = thisThread;try {available.awaitNanos(delay);} finally {if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && queue[0] != null)available.signal();lock.unlock();}
}
siftDown 堆排序中的向下堆操作

由于堆顶元素由于在take()poll()时已经出列而不再存在,堆的结构已经不完整,需调整堆结构。

  1. 把堆顶作为当前节点向下堆化
  2. 取当前节点的较小子节点child与队列最后元素(最后元素为入参的参数key,调用siftDown()时传入)比较
  3. 如果子节点child小于key元素,子节点child与父节点交换,当前节点变为child,然后回到步骤2直至子节点child大于key元素
  4. key元素放在child节点处
/*** 堆排序中的下沉操作* 仅当持有锁的时候才会调用。*/
private void siftDown(int k, RunnableScheduledFuture<?> key) {int half = size >>> 1;  //相当于size/2while (k < half) {   //half之后都是叶子节点,不需要遍历小堆的叶子int child = (k << 1) + 1;RunnableScheduledFuture<?> c = queue[child];int right = child + 1;//如果右节点小于左节点,取右节点作为比较和交换的节点if (right < size && c.compareTo(queue[right]) > 0)c = queue[child = right];if (key.compareTo(c) <= 0)break;queue[k] = c;setIndex(c, k);k = child;}queue[k] = key;setIndex(key, k);
}

思考一个问题:为什么offer()使用的锁方法是lock(),而take()方法使用的锁是lockInterruptibly()

首先来理清一下这两者的区别:

  • lock(): 这是最基本的获取锁的方法。如果调用 lock() 时锁已经被其他线程持有,则当前线程会一直阻塞,直到获取到锁为止。如果在阻塞的过程中当前线程被中断(通过 Thread.interrupt() 方法),它将继续等待锁,不会响应中断,直到获取到锁为止。

  • lockInterruptibly:这个方法也是用来获取锁的,但是它允许在等待锁的过程中响应中断。如果调用 lockInterruptibly() 时锁已经被其他线程持有,则当前线程会被阻塞,但是如果在等待过程中被中断,则会抛出 InterruptedException 异常,此时可以根据实际需求进行相应的处理。

使用offer方法时为了保证提交的任务入列,肯定是要一直阻塞直到获得锁。

使用take()方法取数据时若线程处于竞争锁的时候(此时没有在执行任务,是空闲的工作线程),若调用线程池的shutdown()方法,线程池中断空闲线程时也能抛出 InterruptedException;但如果用的是lock()方法,会导致取得锁后一直在available.await(),除了使用kill -9强杀进程外永远无法回收线程。

leader线程用于

7. ScheduledFutureTask

计划线程池会把RunnableCallable的类型的任务包装成ScheduledFutureTask再放进任务队列,这个包装类是实现周期性重复运行任务的关键

ScheduledFutureTask类图

在这里插入图片描述

定义

private class ScheduledFutureTask<V>extends FutureTask<V> implements RunnableScheduledFuture<V> {private final long sequenceNumber;private long time; //设定的执行时间nanoTime//0:不重复执行//正值: 对应FixedRate//负值: 对应FixedDelayprivate final long period; /** The actual task to be re-enqueued by reExecutePeriodic */RunnableScheduledFuture<V> outerTask = this;ScheduledFutureTask(Runnable r, V result, long ns) {super(r, result);this.time = ns;this.period = 0;this.sequenceNumber = sequencer.getAndIncrement();}ScheduledFutureTask(Runnable r, V result, long ns, long period) {super(r, result);this.time = ns;this.period = period;this.sequenceNumber = sequencer.getAndIncrement();}}//省略代码……
}

run方法

要实现周期性重复运行任务的功能只需把任务重复放进等待队列, worker线程会从队列里面取任务来运行。
由以下代码可知计算任务线程池ScheduledThreadPoolExecutor能实现定时重复运行任务的核心就是当次任务运行完毕时把任务重新加入等待队列。

    private class ScheduledFutureTask<V>extends FutureTask<V> implements RunnableScheduledFuture<V> {//省略代码……public void run() {boolean periodic = isPeriodic();if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();  //如果不是定期任务,运行run方法后结束else if (ScheduledFutureTask.super.runAndReset()) { //如果是定期任务,运行run方法,重设置状态setNextRunTime();//把任务重新加入等待队列reExecutePeriodic(outerTask);}}private void setNextRunTime() {long p = period;if (p > 0)time += p;elsetime = triggerTime(-p);}}void reExecutePeriodic(RunnableScheduledFuture<?> task) {if (canRunInCurrentRunState(true)) {super.getQueue().add(task); //把任务重新加入等待队列if (!canRunInCurrentRunState(true) && remove(task))task.cancel(false);elseensurePrestart();}}

compareTo方法

compareTo()方法比较两个任务的私有变量time的大小,用于在队列中进行有序排序。

  • 当前任务执行时间 先于(小于) 要比较的任务的执行时间,返回-1
  • 当前任务执行时间 后于(大于) 要比较的任务的执行时间,返回1

从上面可以得出结论:

  • a.compareTo(b) < 0:a小于b,即a的执行时间在前
  • a.compareTo(b) > 0:a大于b,即a的执行时间在后
    public int compareTo(Delayed other) {if (other == this) // compare zero if same objectreturn 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;elsereturn 1;}long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;}

cancel方法

cancel方法中的参数mayInterruptIfRunning表示是否中断正在运行的线程,
在计划线程池ScheduledThreadPoolExecutor内部的使用中,mayInterruptIfRunning都为false,少于中断线程这个操作。

cancel(false)方法里,尝试将新任务的状态转为CANCELLED,转为CANCELLED成功后,如果removeOnCancel为true则把任务从队列中移除,如果不是新任务(状态为NEW)则表示任务已经开始运行,无法取消,在FutureTask类的run()方法里如果状态为非NEW的时候会停止执行任务。

removeOnCancel默认为false,也就是不会从队列出移除已取消的任务,但当已取消的任务在线程中运行时,由于状态为CANCELLED会马上返回,不会继续执行。

cancel方法的流程图如下

在这里插入图片描述

源码:

    public boolean cancel(boolean mayInterruptIfRunning) {boolean cancelled = super.cancel(mayInterruptIfRunning);if (cancelled && removeOnCancel && heapIndex >= 0)remove(this);return cancelled;}//继承自FutureTask    public boolean cancel(boolean mayInterruptIfRunning) {//尝试将状态为NEW的任务的转为CANCELLED状态if (!(state == NEW &&UNSAFE.compareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))return false;try {    // in case call to interrupt throws exceptionif (mayInterruptIfRunning) {try {Thread t = runner;if (t != null)t.interrupt();} finally { // final stateUNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);}}} finally {//能执行到这里的都是将任务状态成功转为CANCELLED的,//逐个通知等待结果的线程任务已经被取消finishCompletion(); }return true;}

8. 任务在计划线程线执行过程

任务在提交后最终会在工作线程中执行,worker工作线程run()方法的调用栈比基础线程池的简单Runnable任务要嵌套了几层,理清后对整体流程有更好的理解。

worker工作线程启动后的调用栈

在这里插入图片描述

调用栈执行顺序

  1. Thread.run()
  2. ThreadPoolExecutor$Worker#run();
  3. ThreadPoolExecutor$Worker#runWorker()
  4. ScheduledThreadPoolExecutor$ScheduledFutureTask#run()
  5. FutureTask#runAndReset()
  6. FutureTask#call()
  7. Executors$RunnableAdapter#call()
  8. 提交的执行任务run()方法

可以看到工作线程的run()方法被包裹了几层,每个被包裹层增加了特定功能:

调用栈备注
Worker#run()工作线程
ScheduledFutureTask#run()实现 周期性重复 运行
FutureTask#runAndReset()重置FutureTask的结果状态以便不影响后续重复执行
FutureTask#call()实现了Callable
Executors$RunnableAdapter#call()适配器,将Runnable转为Callable
提交的任务run()方法真正提交的任务

任务从提交开始在计划线程线执行过程的时序图如下:

在这里插入图片描述

任务从提交开始在计划线程线执行过程的简要代码如下:

public class ScheduledThreadPoolExecutor {//工作线程类,初始化的时候会创建线程,线程的运行的Runnable为this(当前worker实例)private final class Worker implements Runnable {final Thread thread;Runnable firstTask;Worker(Runnable firstTask) {this.firstTask = firstTask;//创建线程, 线程的运行的Runnable为this(当前worker实例)this.thread = getThreadFactory().newThread(this);}//线程运行的run方法public void run() {runWorker(this);  }}//1. 提交任务,由计划线程池进行任务调度public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {//省略代码……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);  //调用2return t;}//2. 把任务放入队列private void delayedExecute(RunnableScheduledFuture<?> task) {if (isShutdown())reject(task);else {super.getQueue().add(task);if (//省略代码……)//省略代码……elseensurePrestart();  //调用步骤3确保线程已预热}}//3. 预热,确保工作线程已启动  void ensurePrestart() {int wc = workerCountOf(ctl.get());if (wc < corePoolSize)addWorker(null, true);else if (wc == 0)addWorker(null, false);}//4. 增加并启动worker工作线程private boolean addWorker(Runnable firstTask, boolean core) {Worker w = null;try {//省略代码……w = new Worker(firstTask); //2.1 新建worker实例final Thread t = w.thread;if (t != null) {try {//省略代码……workers.add(w);   //2.2 放入集合用于管理}if (workerAdded) {//2.3 启动worker线程,//    线程启动后运行worker的run()方法,//    worker的run()会运行runWorker(this)方法,//    所以线程启动后实际运行的是runWorker(this)方法;t.start();        }}//省略代码……}return workerStarted;}//5. 线程运行循环体final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;//省略代码……try {//核心循环体: 执行第一个任务,完成后循环地从队列取任务执行while (task != null || (task = getTask()) != null) {//5.1 通过getTask()取任务//省略代码……try {try {task.run();  //5.2 运行任务,任务的类型为ScheduledFutureTask} catch (RuntimeException x) {thrown = x; throw x;}//省略代码……}//省略代码……}}}private class ScheduledFutureTask<V>extends FutureTask<V> implements RunnableScheduledFuture<V> {//6. 运行任务,计算任务下个周期运行的时间缀,任务重新放回队列public void run() {boolean periodic = isPeriodic();if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();else if (ScheduledFutureTask.super.runAndReset()) { //6.1 runAndReset()setNextRunTime();               //6.2 计算任务下个周期运行的时间缀reExecutePeriodic(outerTask);   //6.3 任务重新放回队列}}//6.1 该方法继承自父类FutureTask, 重置状态的是任务是否已完成protected boolean runAndReset() {//省略代码……boolean ran = false;int s = state;try {Callable<V> c = callable;if (c != null && s == NEW) {try {c.call(); // don't set resultran = true;} catch (Throwable ex) {setException(ex);}}} finally {//省略代码……}return ran && s == NEW;}}//6.2 计算下个周期运行的时间缀private void setNextRunTime() {long p = period;if (p > 0)time += p;elsetime = triggerTime(-p);}//6.3 任务重新放回队列, 线程池的线程会从队列取任执行(步骤5.1)void reExecutePeriodic(RunnableScheduledFuture<?> task) {if (canRunInCurrentRunState(true)) {super.getQueue().add(task);//省略代码……}}}

9. 计划线程池ScheduledThreadPoolExecutor的关闭

计划线程池的关闭继承自线程池ThreadPoolExecutor的shutdown()shutdownNow(),保留了这两个方法的全部功能。(如果对ThreadPoolExecutor还不是很熟悉,可以阅读一下这篇文章:
Java线程池ThreadPoolExecutor运行机制和源码解析。)

另外,ThreadPoolExecutor给计划线程池ScheduledThreadPoolExecutor预留的调用钩子onShutdown(),用于根据executeExistingDelayedTasksAfterShutdowncontinueExistingPeriodicTasksAfterShutdown这两个策略判断是否继续运行延迟和周期性任务。

和线程池不同的是计划线程池在onShutdown()根据条件对延迟任务和周期性任务进行取消操作。

  • executeExistingDelayedTasksAfterShutdown:是否在shutdown后继续运行延迟任务
  • continueExistingPeriodicTasksAfterShutdown:是否在shutdown后继续运行周期性任务

onShutdown()源码

@Override void onShutdown() {BlockingQueue<Runnable> q = super.getQueue();boolean keepDelayed =getExecuteExistingDelayedTasksAfterShutdownPolicy();boolean keepPeriodic =getContinueExistingPeriodicTasksAfterShutdownPolicy();if (!keepDelayed && !keepPeriodic) {//如果保留延迟任务和保留周期性任务都为false, 清空队列for (Object e : q.toArray())if (e instanceof RunnableScheduledFuture<?>)((RunnableScheduledFuture<?>) e).cancel(false);q.clear();}else {// Traverse snapshot to avoid iterator exceptionsfor (Object e : q.toArray()) {if (e instanceof RunnableScheduledFuture) {RunnableScheduledFuture<?> t =(RunnableScheduledFuture<?>)e;if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) ||//满足以下条件的一个则把任务从队列移除并取消任务//1.是周期性任务但设置了不保留周期性任务//2.一次性延迟任务但设置了不保留延迟任务//3.任务已取消t.isCancelled()) { // also remove if already cancelledif (q.remove(t))t.cancel(false);}}}}tryTerminate();}
}

结语

本文对java计划线程池ScheduledThreadPoolExecutor的分析就到这里。

如有不正,欢迎指出。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/704142.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【C++】---内存管理new和delete详解

一、C/C内存分布 C/C内存被分为6个区域&#xff1a; &#xff08;1&#xff09; 内核空间&#xff1a;存放内核代码和环境变量。 &#xff08;2&#xff09;栈区&#xff1a;向下增长&#xff08;存放非静态局部变量&#xff0c;函数参数&#xff0c;返回值等等&#xff09; …

OCPP 1.6 接入实现文档

一、简介 OCPP&#xff08;Open Charge Point Protocol&#xff09;是一个开放的通信协议&#xff0c;用于充电站&#xff08;Charge Point&#xff09;与中央系统&#xff08;Central System&#xff0c;如充电站管理系统或服务提供商平台&#xff09;之间的通讯。本篇文档将…

纽约纳斯达克大屏投放受众群体有哪些-大舍传媒

纽约纳斯达克大屏投放受众群体有哪些-大舍传媒 1. 纳斯达克大屏的概述 纳斯达克大屏是全球金融市场中最出名的电子交易平台之一。作为一个重要的金融信息传递渠道&#xff0c;纳斯达克大屏吸引了来自全球的投资者的目光。在这个巨大的投放平台上&#xff0c;大舍传媒希望为客…

【Java】线程池总结——干货满满!!!

目录 一 使用线程池的好处 二 Executor 框架 2.1 简介 2.2 Executor 框架结构(主要由三大部分组成) 1) 任务(Runnable /Callable) 2) 任务的执行(Executor) 3) 异步计算的结果(Future) 2.3 Executor 框架的使用示意图 三 (重要)ThreadPoolExecutor 类简单介绍 3.1 Thr…

如何使用Axure RP制作web页面并实现无公网ip远程访问——“cpolar内网穿透”

文章目录 前言1.在AxureRP中生成HTML文件2.配置IIS服务3.添加防火墙安全策略4.使用cpolar内网穿透实现公网访问4.1 登录cpolar web ui管理界面4.2 启动website隧道4.3 获取公网URL地址4.4. 公网远程访问内网web站点4.5 配置固定二级子域名公网访问内网web站点4.5.1创建一条固定…

Vue知识学习

Vue 是什么&#xff1f; 概念&#xff1a;Vue 是一个用于构建用户界面的渐进式框架 Vue 的两种使用方式: ① Vue 核心包开发 场景:局部 模块改造 ② Vue 核心包& Vue插件工程化开发 场景:整站开发 创建Vue 实例&#xff0c;初始化渲染的核心步骤: 1.准备容器 2.引包(官…

基于java在线调查表单系统

基于java在线调查表单系统 一、演示效果二、特性汇总三、下载链接 一、演示效果 二、特性汇总 多种技术方案&#xff0c;满足不同的技术选型需求完善的浏览器兼容、保证传统客户也能正常使用部署简单&#xff0c;一行命令完成部署更新方便&#xff0c;直接替换原安装文件不用担…

F2图例封装 - BarAndLine

基于vue3 和 F2 3.5.0 <template><div :style"{minHeight: ${height}px,width: 100% }" ref"container"><canvas v-show"showChart" ref"canvas" :id"chartId" class"chart-canval"></canva…

【stm32】hal库-双通道ADC采集

【stm32】hal库-双通道ADC采集 CubeMX图形化配置 程序编写 /* USER CODE BEGIN PV */ #define BATCH_DATA_LEN 1 uint32_t dmaDataBuffer[BATCH_DATA_LEN]; /* USER CODE END PV *//* USER CODE BEGIN 2 */lcd_init();lcd_show_str(10, 10, 24, "Demo14_4:ADC1 ADC2 S…

Python 实现 BRAR 指标计算(情绪指标):股票技术分析的利器系列(11)

Python 实现 BRAR 指标计算&#xff08;情绪指标&#xff09;&#xff1a;股票技术分析的利器系列&#xff08;11&#xff09; 介绍算法公式 代码rolling函数介绍核心代码计算BR计算AR 完整代码 介绍 BRAR 是一种情绪指标&#xff0c;用于衡量特定金融市场中的买卖情绪。它代表…

试一下newb,还是有错误呀

解题&#xff1a;原式&#xff1d; 2. 在递增的等比数列 ( a n ) (a_n) (an​)中&#xff0c;若 ( a 3 − a 1 5 2 ) (a_3 - a_1 \frac{5}{2}) (a3​−a1​25​), ( a 2 3 ) (a_2 3) (a2​3), 则公比 (q) A. ( 4 3 ) ( \frac{4}{3} ) (34​) B. ( 3 2 ) ( \frac{3}{2} …

旧物回收小程序开发,开启绿色生活新篇章

随着科技的发展和人们生活水平的提高&#xff0c;物质生活的丰富带来了大量的废弃物。如何合理处理这些废弃物&#xff0c;实现资源的再利用&#xff0c;已成为社会关注的焦点。旧物回收小程序的开发与应用&#xff0c;为这一问题提供了有效的解决方案。本文将探讨旧物回收小程…

软考47-上午题-【数据库】-数据查询语言DQL2

一、聚合函数 聚合函数实现数据统计的功能&#xff0c;返回一个单一的值。聚合函数一般与select语句的group by子句一起使用。 示例&#xff1a; 二、数据分组-group by 聚合函数加上group by子句进行分组。 通常一个聚合函数的作用范围是满足where子句中指定条件的记录&…

Atcoder ABC341 C - Takahashi Gets Lost

Takahashi Gets Lost&#xff08;高桥迷路了&#xff09; 时间限制&#xff1a;3s 内存限制&#xff1a;1024MB 【原题地址】 所有图片源自Atcoder&#xff0c;题目译文源自脚本Atcoder Better! 点击此处跳转至原题 【问题描述】 【输入格式】 【输出格式】 【样例1】 【…

vscode输入英文时字体之间的间隔突然变大,似中文

vscode输入英文时字体之间的间隔突然变大&#xff0c;似中文 主要原因&#xff1a; 是由于输入法变成全角模式了。原因可能是不小心按了 shift空格键快捷键造成的。 正常情况&#xff0c;全角就是字母和数字等与汉字占等宽位置的字。 半角就是ASCII方式的字符&#xff0c;在没…

并发编程线程安全性之可见性有序性

可见性 可见性: 就是说一个线程对共享变量的修改&#xff0c;另一个线程能够立刻看到 通俗点说&#xff0c;就是两个线程共享一个变量&#xff0c;无论哪一个线程修改了这个变量&#xff0c;另外一个线程都能够立刻看到上一个线程对这个变量的修改 产生线程安全问题的原因 计…

桶装水配送小程序有什么功能 怎么制作

桶装水配送小程序是一种方便快捷的订水服务工具&#xff0c;可以帮助用户轻松订购桶装水&#xff0c;并实时跟踪订单配送情况。下面具体介绍送水小程序的功能。 1. 客户订水功能&#xff1a;用户可以通过小程序轻松选择需要的桶装水数量和配送时间&#xff0c;填写联系信息和地…

leet hot 100-7 无重复字符的最长子串

无重复字符的最长子串 原题链接思路代码 原题链接 leet hot 100-7 3. 无重复字符的最长子串 思路 声明一个无序map容器&#xff0c;容器里面记录着窗口里面字符的数量 &#xff0c;如果字符的数量大于1 遍历到s[i]的时候 数量1 此时如果map里面相关的数量大于1 说明窗口里面…

ocr识别tesseract.js本地复现

来源&#xff1a; https://github.com/naptha/tesseract.js chatgpt今天帮倒忙&#xff0c;一直给一些旧的东西&#xff0c;代码就老报错&#xff0c;最后还是我出面看看log和err调了一下&#xff0c;还的是我啊 复现效果 这个挺好复现的&#xff0c;用的英文模式比中文识别…

阿里云服务器ECS u1实例性能如何?199元一年

阿里云服务器ECS u1实例&#xff0c;2核4G&#xff0c;5M固定带宽&#xff0c;80G ESSD Entry盘优惠价格199元一年&#xff0c;性能很不错&#xff0c;CPU采用Intel Xeon Platinum可扩展处理器&#xff0c;购买限制条件为企业客户专享&#xff0c;实名认证信息是企业用户即可&a…