深读源码-java线程系列之线程池深入解析——定时任务执行流程

本文主要是介绍深读源码-java线程系列之线程池深入解析——定时任务执行流程,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

简介

前面我们一起学习了普通任务、未来任务的执行流程,今天我们再来学习一种新的任务——定时任务。

定时任务是我们经常会用到的一种任务,它表示在未来某个时刻执行,或者未来按照某种规则重复执行的任务。

问题

(1)如何保证任务是在未来某个时刻才被执行?

(2)如何保证任务按照某种规则重复执行?

来个栗子

创建一个定时线程池,用它来跑四种不同的定时任务。

public class ThreadPoolTest03 {public static void main(String[] args) throws ExecutionException, InterruptedException {// 创建一个定时线程池ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);System.out.println("start: " + System.currentTimeMillis());// 执行一个无返回值任务,5秒后执行,只执行一次scheduledThreadPoolExecutor.schedule(() -> {System.out.println("spring: " + System.currentTimeMillis());}, 5, TimeUnit.SECONDS);// 执行一个有返回值任务,5秒后执行,只执行一次ScheduledFuture<String> future = scheduledThreadPoolExecutor.schedule(() -> {System.out.println("inner summer: " + System.currentTimeMillis());return "outer summer: ";}, 5, TimeUnit.SECONDS);// 获取返回值System.out.println(future.get() + System.currentTimeMillis());// 按固定频率执行一个任务,每2秒执行一次,1秒后执行// 任务开始时的2秒后scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {System.out.println("autumn: " + System.currentTimeMillis());LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));}, 1, 2, TimeUnit.SECONDS);// 按固定延时执行一个任务,每延时2秒执行一次,1秒执行// 任务结束时的2秒后scheduledThreadPoolExecutor.scheduleWithFixedDelay(() -> {System.out.println("winter: " + System.currentTimeMillis());LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));}, 1, 2, TimeUnit.SECONDS);}
}

定时任务总体分为四种:

(1)未来执行一次的任务,无返回值;

(2)未来执行一次的任务,有返回值;

(3)未来按固定频率重复执行的任务;

(4)未来按固定延时重复执行的任务;

本文主要以第三种为例进行源码解析。

scheduleAtFixedRate()方法

提交一个按固定频率执行的任务。

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();// 将普通任务装饰成ScheduledFutureTaskScheduledFutureTask<Void> sft =new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));// 钩子方法,给子类用来替换装饰task,这里认为t==sftRunnableScheduledFuture<Void> t = decorateTask(command, sft);sft.outerTask = t;// 延时执行delayedExecute(t);return t;
}

可以看到,这里的处理跟未来任务类似,都是装饰成另一个任务,再拿去执行,不同的是这里交给了delayedExecute()方法去执行,这个方法是干嘛的呢?

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();}
}
void ensurePrestart() {int wc = workerCountOf(ctl.get());// 创建工作线程// 注意,这里没有传入firstTask参数,因为上面先把任务扔到队列中去了// 另外,没用上maxPoolSize参数,所以最大线程数量在定时线程池中实际是没有用的if (wc < corePoolSize)addWorker(null, true);else if (wc == 0)addWorker(null, false);
}

到这里就结束了?!

实际上,这里只是控制任务能不能被执行,真正执行任务的地方在任务的run()方法中。

还记得上面的任务被装饰成了ScheduledFutureTask类的实例吗?所以,我们只要看ScheduledFutureTask的run()方法就可以了。

ScheduledFutureTask类的run()方法

定时任务执行的地方。

public void run() {// 是否重复执行boolean periodic = isPeriodic();// 线程池状态判断if (!canRunInCurrentRunState(periodic))cancel(false);// 一次性任务,直接调用父类的run()方法,这个父类实际上是FutureTask// 这里我们不再讲解,有兴趣的同学看看上一章的内容else if (!periodic)ScheduledFutureTask.super.run();// 重复性任务,先调用父类的runAndReset()方法,这个父类也是FutureTask// 本文主要分析下面的部分else if (ScheduledFutureTask.super.runAndReset()) {// 设置下次执行的时间setNextRunTime();// 重复执行reExecutePeriodic(outerTask);}
}

可以看到,对于重复性任务,先调用FutureTask的runAndReset()方法,再设置下次执行的时间,最后再调用reExecutePeriodic()方法。

FutureTask的runAndReset()方法与run()方法类似,只是其任务运行完毕后不会把状态修改为NORMAL,有兴趣的同学点进源码看看。

再来看看reExecutePeriodic()方法。

void reExecutePeriodic(RunnableScheduledFuture<?> task) {// 线程池状态检查if (canRunInCurrentRunState(true)) {// 再次把任务扔到任务队列中super.getQueue().add(task);// 再次检查线程池状态if (!canRunInCurrentRunState(true) && remove(task))task.cancel(false);else// 保证工作线程足够ensurePrestart();}
}

到这里是不是豁然开朗了,原来定时线程池执行重复任务是在任务执行完毕后,又把任务扔回了任务队列中。

重复性的问题解决了,那么,它是怎么控制任务在某个时刻执行的呢?

OK,这就轮到我们的延时队列登场了。

DelayedWorkQueue内部类

我们知道,线程池执行任务时需要从任务队列中拿任务,而普通的任务队列,如果里面有任务就直接拿出来了,但是延时队列不一样,它里面的任务,如果没有到时间也是拿不出来的,这也是前面分析中一上来就把任务扔进队列且创建Worker没有传入firstTask的原因。

说了这么多,它到底是怎么实现的呢?

其实,延时队列我们在前面都详细分析过,想看完整源码分析的可以看看之前的《深读源码-java集合之DelayQueue源码分析》。

延时队列内部是使用“堆”这种数据结构来实现的,有兴趣的同学可以看看之前的《什么是堆》。

我们这里只拿一个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);// 如果小于等于0,说明这个任务到时间了,可以从队列中出队了if (delay <= 0)// 出队,然后堆化return finishPoll(first);// 还没到时间first = null;// 如果前面有线程在等待,直接进入等待if (leader != null)available.await();else {// 当前线程作为leaderThread thisThread = Thread.currentThread();leader = thisThread;try {// 等待上面计算的延时时间,再自动唤醒available.awaitNanos(delay);} finally {// 唤醒后再次获得锁后把leader再置空if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && queue[0] != null)// 相当于唤醒下一个等待的任务available.signal();// 解锁lock.unlock();}
}

大致的原理是,利用堆的特性获取最快到时间的任务,即堆顶的任务:

(1)如果堆顶的任务到时间了,就让它从队列中了队;

(2)如果堆顶的任务还没到时间,就看它还有多久到时间,利用条件锁等待这段时间,待时间到了后重新走(1)的判断;

这样就解决了可以在指定时间后执行任务。

其它

其实,ScheduledThreadPoolExecutor也是可以使用execute()或者submit()提交任务的,只不过它们会被当成0延时的任务来执行一次。

public void execute(Runnable command) {schedule(command, 0, NANOSECONDS);
}
public <T> Future<T> submit(Callable<T> task) {return schedule(task, 0, NANOSECONDS);
}

总结

实现定时任务有两个问题要解决,分别是指定未来某个时刻执行任务、重复执行。

(1)指定某个时刻执行任务,是通过延时队列的特性来解决的;

(2)重复执行,是通过在任务执行后再次把任务加入到队列中来解决的。

彩蛋

到这里基本上普通的线程池的源码解析就结束了,这种线程池是比较经典的实现方式,整体上来说,效率相对不是特别高,因为所有的工作线程共用同一个队列,每次从队列中取任务都要加锁解锁操作。

那么,能不能给每个工作线程配备一个任务队列呢,在提交任务的时候就把任务分配给指定的工作线程,这样在取任务的时候就不需要频繁的加锁解锁了。

答案是肯定的,下一章我们一起来看看这种基于“工作窃取”理论的线程池——ForkJoinPool。


原文链接:https://www.cnblogs.com/tong-yuan/p/11801757.html

这篇关于深读源码-java线程系列之线程池深入解析——定时任务执行流程的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


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

相关文章

Java中Arrays类和Collections类常用方法示例详解

《Java中Arrays类和Collections类常用方法示例详解》本文总结了Java中Arrays和Collections类的常用方法,涵盖数组填充、排序、搜索、复制、列表转换等操作,帮助开发者高... 目录Arrays.fill()相关用法Arrays.toString()Arrays.sort()A

Spring Boot Maven 插件如何构建可执行 JAR 的核心配置

《SpringBootMaven插件如何构建可执行JAR的核心配置》SpringBoot核心Maven插件,用于生成可执行JAR/WAR,内置服务器简化部署,支持热部署、多环境配置及依赖管理... 目录前言一、插件的核心功能与目标1.1 插件的定位1.2 插件的 Goals(目标)1.3 插件定位1.4 核

如何使用Lombok进行spring 注入

《如何使用Lombok进行spring注入》本文介绍如何用Lombok简化Spring注入,推荐优先使用setter注入,通过注解自动生成getter/setter及构造器,减少冗余代码,提升开发效... Lombok为了开发环境简化代码,好处不用多说。spring 注入方式为2种,构造器注入和setter

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4

Java堆转储文件之1.6G大文件处理完整指南

《Java堆转储文件之1.6G大文件处理完整指南》堆转储文件是优化、分析内存消耗的重要工具,:本文主要介绍Java堆转储文件之1.6G大文件处理的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言文件为什么这么大?如何处理这个文件?分析文件内容(推荐)删除文件(如果不需要)查看错误来源如何避

SpringBoot整合Dubbo+ZK注册失败的坑及解决

《SpringBoot整合Dubbo+ZK注册失败的坑及解决》使用Dubbo框架时,需在公共pom添加依赖,启动类加@EnableDubbo,实现类用@DubboService替代@Service,配... 目录1.先看下公共的pom(maven创建的pom工程)2.启动类上加@EnableDubbo3.实

SpringBoot整合(ES)ElasticSearch7.8实践

《SpringBoot整合(ES)ElasticSearch7.8实践》本文详细介绍了SpringBoot整合ElasticSearch7.8的教程,涵盖依赖添加、客户端初始化、索引创建与获取、批量插... 目录SpringBoot整合ElasticSearch7.8添加依赖初始化创建SpringBoot项

JAVA覆盖和重写的区别及说明

《JAVA覆盖和重写的区别及说明》非静态方法的覆盖即重写,具有多态性;静态方法无法被覆盖,但可被重写(仅通过类名调用),二者区别在于绑定时机与引用类型关联性... 目录Java覆盖和重写的区别经常听到两种话认真读完上面两份代码JAVA覆盖和重写的区别经常听到两种话1.覆盖=重写。2.静态方法可andro

SpringBoot中六种批量更新Mysql的方式效率对比分析

《SpringBoot中六种批量更新Mysql的方式效率对比分析》文章比较了MySQL大数据量批量更新的多种方法,指出REPLACEINTO和ONDUPLICATEKEY效率最高但存在数据风险,MyB... 目录效率比较测试结构数据库初始化测试数据批量修改方案第一种 for第二种 case when第三种

Java docx4j高效处理Word文档的实战指南

《Javadocx4j高效处理Word文档的实战指南》对于需要在Java应用程序中生成、修改或处理Word文档的开发者来说,docx4j是一个强大而专业的选择,下面我们就来看看docx4j的具体使用... 目录引言一、环境准备与基础配置1.1 Maven依赖配置1.2 初始化测试类二、增强版文档操作示例2.