elastic-job 定时任务集成
文章目录
- 一、基本使用
- 第一步添加依赖：
- 第二步:增加Zookeeper注册中心的配置
- 第三步:开启Elastic-Job自动配置
- 第四步 配置任务
- 二、扩展功能
- 1.想在配置文件设置任务参数怎么写?
- 2.每次任务执行日志如何记录?
- 3.想执行业务脚本任务如何操作?
- 4.任务执行前后如何添加监听事件?
- 5.任务业务执行失败如何自定义异常处理?
- 6.动态作业怎么实现?
“前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站。”
一、基本使用
elastic-job的任务类型分为三种：
- a. Simple类型作业,意为简单实现，未经任何封装的类型。需实现SimpleJob接口。该接口仅提供单一方法用于覆盖，此方法将定时执行。与Quartz原生接口相似，但提供了弹性扩缩容和分片等功能。
- b. Dataflow类型用于处理数据流，需实现DataflowJob接口。该接口提供2个方法可供覆盖，分别用于抓取(fetchData)和处理(processData)数据。
- c.Script类型作业意为脚本类型作业，支持shell，python，perl等所有类型脚本。只需通过控制台或代码配置scriptCommandLine即可，无需编码。执行脚本路径可包含参数，参数传递完毕后，作业框架会自动追加最后一个参数为作业运行时信息。
本文档提供的jar包是对elastic-job进行了二次封装，封装成了springboot的starter启动器，使用更加简单。后续开发引入我们自定义的starter完成定时任务功能即可。
先以最简单的SimpleJob类型说明如何使用集成：
第一步添加依赖：
<!--elastic-job的starter集成-->
<dependency><groupId>com.wuzheng</groupId><artifactId>spring-boot-elastic-job-starter</artifactId><version>1.0.4</version>
</dependency>
点击此处查看starter源码，可以自己下载到本地或私服，引入依赖
第二步:增加Zookeeper注册中心的配置
elastic.job.zk.serverLists=192.168.0.121:2181,192.168.0.122:2181,192.168.0.129:2181
elastic.job.zk.namespace=yourself namespace
Zookeeper配置的前缀是elastic.job.zk，详细的属性配置请查看ZookeeperProperties
第三步:开启Elastic-Job自动配置
开启自动配置只需要在Spring Boot的启动类上增加@EnableElasticJob注解
import java.util.concurrent.CountDownLatch;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import com.cxytiandi.elasticjob.annotation.EnableElasticJob;/*** ElasticJob Spring Boot集成案例* * @author hfl**/
@SpringBootApplication
@EnableElasticJob
public class JobApplication {public static void main(String[] args) {SpringApplication.run(JobApplication.class, args);}}
第四步 配置任务
@ElasticJobConf(name = "MySimpleJob", cron = "0/10 * * * * ?", shardingItemParameters = "0=0,1=1", description = "简单任务")
public class MySimpleJob implements SimpleJob {public void execute(ShardingContext context) {String shardParamter = context.getShardingParameter();System.out.println("分片参数："+shardParamter);int value = Integer.parseInt(shardParamter);for (int i = 0; i < 3; i++) {if (i % 2 == value) {String time = new SimpleDateFormat("HH:mm:ss").format(new Date());System.out.println(time + ":开始执行简单任务" + i);}}}}
任务的配置只需要在任务类上增加一个ElasticJobConf注解，注解中有很多属性，这些属性都是任务的配置，详细的属性配置如下链接：
ElasticJobConf.java
测试下，启动自己的任务所在微服务，看下效果
自此，最简单的小demo其实已经完成了，但是实际中还有一些高级功能需要了解，在此也一并说明下：
二、扩展功能
扩展包括:
- 1.想在配置文件设置任务参数怎么写?
- 2.每次任务执行日志如何记录?
- 3.想执行业务脚本任务如何操作?
- 4.任务执行前后如何添加监听事件?
- 5.任务业务执行失败如何自定义异常处理?
- 6.动态作业怎么实现?
下面具体说明：
1.想在配置文件设置任务参数怎么写?
这个问题即application.properties中如何配置任务信息?
使用注解是比较方便，但是在属性文件中配置任务的参数信息更灵活，也便于以后使用appolo或者necos等工具进行集中配置文件管理。
我们可以同时指定注解方法，也可以同时在属性文件中设置，但是当属性文件中配置了任务的信息，优先级就比注解中的高。具体做法如下:
首先还是在任务类上加@ElasticJobConf(name = “MySimpleJob”)注解，只需要增加一个name即可，任务名是唯一的。
剩下的配置都可以在属性文件中进行配置，格式为elastic.job.任务名.配置属性=属性值
elastic.job.MySimpleJob.cron=0/10 * * * * ?
elastic.job.MySimpleJob.overwrite=true
elastic.job.MySimpleJob.shardingTotalCount=1
elastic.job.MySimpleJob.shardingItemParameters=0=0,1=1
elastic.job.MySimpleJob.jobParameter=test
elastic.job.MySimpleJob.failover=true
elastic.job.MySimpleJob.misfire=true
elastic.job.MySimpleJob.description=simple job
elastic.job.MySimpleJob.monitorExecution=false
elastic.job.MySimpleJob.listener=com.cxytiandi.job.core.MessageElasticJobListener
elastic.job.MySimpleJob.jobExceptionHandler=com.cxytiandi.job.core.CustomJobExceptionHandler
elastic.job.MySimpleJob.disabled=true
2.每次任务执行日志如何记录?
elastic-job已经为我们提供了事件追踪数据源功能，可以将每一次的任务执行情况，任务执行轨迹，任务执行历史持久化到数据库或者其他日志收集框架中，下面演示如何保存到数据库中：
事件追踪功能在注解中也只需要配置eventTraceRdbDataSource=你的数据源 就可以使用了，数据源用什么连接池无限制，唯一需要注意的一点是你的数据源必须在spring-boot-elastic-job-starter之前创建，因为spring-boot-elastic-job-starter中依赖了你的数据源，下面我以druid作为连接池来进行讲解。
<dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>1.1.2</version>
</dependency>
配置连接池属性：
spring.datasource.druid.log.url=jdbc:mysql://localhost:3306/event_log
spring.datasource.druid.log.username=root
spring.datasource.druid.log.password=123456
spring.datasource.druid.log.driver-class-name=com.mysql.jdbc.Driver
然后在项目中定义一个配置类，配置连接池，手动配置的原因是连接池可以在elastic-job-starter之前被初始化。
@Configuration
public class BeanConfig {/*** 任务执行事件数据源* @return*/@Bean("datasource")@ConfigurationProperties("spring.datasource.druid.log")public DataSource dataSourceTwo(){return DruidDataSourceBuilder.create().build();}}然后在注解中增加数据源的配置eventTraceRdbDataSource 即可：
注解配置方式如下:
@ElasticJobConf(name = "MySimpleJob", cron = "0/10 * * * * ?", shardingItemParameters = "0=0,1=1", description = "简单任务", eventTraceRdbDataSource = "datasource")
也可以通过官方提供的可视化工具看到任务执行情况
注意: 经测试:
1：经测试: 该功能Mysql数据库的话只支持5.7，Mysql5.8并不会生效
2： 数据库名必须是elastic_job_log 才会自动创建这2张表，其他数据库名请手动添加job_execution_log，job_status_trace_log这2张表（后面附2张表建表sql）
3：zookeepr注册中心必须能够连到mysql
CREATE TABLE `job_execution_log` (`id` varchar(40) NOT NULL,`job_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '作业名称',`task_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务名称,每次作业运行生成新任务',`hostname` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主机名称',`ip` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主机IP',`sharding_item` int NOT NULL COMMENT '分片项',`execution_source` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '作业执行来源。可选值为NORMAL_TRIGGER, MISFIRE, FAILOVER',`failure_cause` varchar(4000) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL COMMENT '执行失败原因',`is_success` int NOT NULL COMMENT '是否执行成功',`start_time` timestamp NULL DEFAULT NULL COMMENT '作业开始执行时间',`complete_time` timestamp NULL DEFAULT NULL COMMENT '作业结束执行时间',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='历史轨迹表';CREATE TABLE `job_status_trace_log` (`id` varchar(40) NOT NULL,`job_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '作业名称',`original_task_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '原任务id',`task_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务id',`slave_id` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '执行作业服务器的名称，Lite版本为服务器的IP地址，Cloud版本为Mesos执行机主键',`source` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务执行源，可选值为CLOUD_SCHEDULER, CLOUD_EXECUTOR, LITE_EXECUTOR',`execution_type` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务执行类型，可选值为NORMAL_TRIGGER, MISFIRE, FAILOVER',`sharding_item` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '分片项集合，多个分片项以逗号分隔',`state` varchar(20) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '任务执行状态，可选值为TASK_STAGING, TASK_RUNNING, TASK_FINISHED, TASK_KILLED, TASK_LOST, TASK_FAILED, TASK_ERROR',`message` varchar(4000) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL COMMENT '相关信息',`creation_time` timestamp NULL DEFAULT NULL COMMENT '记录创建时间',PRIMARY KEY (`id`),KEY `TASK_ID_STATE_INDEX` (`task_id`,`state`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='历史状态表';3.想执行业务脚本任务如何操作?
elastic-job提供了Script任务类型.
由于Script任务的执行逻辑是在具体的脚本中，是通过scriptCommandLine来指定执行脚本的路径。我这边为了统一的去发现项目中的任务列表，还是需要建一个脚本的Java类，加上ElasticJobConf注解，只是不需要写逻辑而已，示例如下：
/*** 脚本任务不需要写逻辑，逻辑在被执行的脚本中，这边只是定义一个任务而已**/
@ElasticJobConf(name = "MyScriptJob")
public class MyScriptJob implements ScriptJob {public void execute(ShardingContext context) {}}配置：
elastic.job.MyScriptJob.cron=0/10 * * * * ?
elastic.job.MyScriptJob.overwrite=true
elastic.job.MyScriptJob.scriptCommandLine=D:\\1.bat
1.bat脚本内容如下:
@echo ------【脚本任务】Sharding Context: %*
4.任务执行前后如何添加监听事件?
首先书写自己的监听事件:
/*** 作业监听器, 执行前后发送钉钉消息进行通知* @author hfl*/
public class MessageElasticJobListener implements ElasticJobListener {@Overridepublic void beforeJobExecuted(ShardingContexts shardingContexts) {String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());String msg = date + " 【五征定时任务-" + shardingContexts.getJobName() + "】任务开始执行====" + JsonUtils.toJson(shardingContexts);
// DingDingMessageUtil.sendTextMessage(msg);System.out.println(msg);}@Overridepublic void afterJobExecuted(ShardingContexts shardingContexts) {String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());String msg = date + " 【五征定时任务-" + shardingContexts.getJobName() + "】任务执行结束====" + JsonUtils.toJson(shardingContexts);
// DingDingMessageUtil.sendTextMessage(msg);System.out.println(msg);}}
然后在配置文件中job参数上添加listener参数:
elastic.job.MySimpleJob.listener=com.wuzheng.job.core.MessageElasticJobListener
运行MySimpleJob看下执行结果:
可以看到任务前后执行了我们自己的逻辑.
5.任务业务执行失败如何自定义异常处理?
和监听器配置一样，第一步书写自定义异常，第二步配置job参数指定自定义异常类。
/*** 自定义异常处理，在任务异常时使用钉钉发送通知* @author hfl*/
public class CustomJobExceptionHandler implements JobExceptionHandler {private Logger logger = LoggerFactory.getLogger(CustomJobExceptionHandler.class);@Overridepublic void handleException(String jobName, Throwable cause) {logger.error(String.format("Job '%s' exception occur in job processing", jobName), cause);DingDingMessageUtil.sendTextMessage("【"+jobName+"】任务异常。" + cause.getMessage());}}
异常参数设置
elastic.job.MySimpleJob.jobExceptionHandler=com.wuzheng.job.core.CustomJobExceptionHandler我们自定义个异常测试下（System.out.println(2/0);）
@ElasticJobConf(name = "MySimpleJob")
public class MySimpleJob implements SimpleJob {@Overridepublic void execute(ShardingContext context) {System.out.println(2/0);String shardParamter = context.getShardingParameter();System.out.println("分片参数："+shardParamter);for (int i = 0; i < 2; i++) {String time = new SimpleDateFormat("HH:mm:ss").format(new Date());System.out.println(time + ":开始执行简单任务" + i);}}}
看下效果：
已经指向了我们自定义的异常的类及其类里面的异常处理逻辑代码：
6.动态作业怎么实现?
以上的定时任务基本都是这个流程:首先配置job参数，然后编写业务job，最后启动定时任务所在的微服务，这样该微服务下的所有定时任务job都统一注册进zookeeper注册中心了，然后自动就会启动这些定时任务。
如果有个定时任务，不想微服务启动的时候让其启用，而是想动态的注册任务参数并控制何时启用。这个需求实现也很简单，我们封装的elastic-job的starter就已提供实现。
自定义的starter提供了restful的2个接口，一个注册任务接口: /job；一个删除任务接口:/job/remove?jobName=任务名。
首先书写自己的任务
package com.wuzheng.job.demo.dynamic;import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;/*** 动态添加任务演示* @author hfl**/
public class DynamicJob implements SimpleJob {@Overridepublic void execute(ShardingContext shardingContext) {// 可以根据JobParameter来对不同的数据进行操作System.out.println(shardingContext.getJobParameter());System.out.println(shardingContext.getShardingParameter());}}
置好之后，启动这个定时任务微服务项目并不会启用这个任务，因为我们没有任务参数，也没在job上加@ElasticJobConf注解，扫描不到这个任务，但是通过REST API可以动态的注册任务，API列表如下：
/job
添加任务是POST请求，数据格式为JSON体提交，格式如下：
{"jobName":"DynamicJob333",
"cron":"0/10 * * * * ?",
"jobType":"SIMPLE",
"jobClass":"com.wuzheng.job.demo.dynamic.DynamicJob",
"jobParameter":"2222222",
"shardingTotalCount":1
}点击查看完整字段地址
注意：jobClass必须事先存在于服务中
使用postman测试:
去控制台参看,显示已经成功注册到zookeeper中心，作业状态正常!
/job/remove 删除任务是GET请求，参数只要任务名称即可，比如：/job/remove?jobName=任务名。可以用于任务完成之后清空注册中心的任务信息。
以上所有demo的源码下载地址请点击：https://gitee.com/hufanglei/elastic-job-learn/tree/master/spring-boot-elastic-job-example。
个人微信公众号：
搜索： 怒放de每一天
不定时推送相关文章，期待和大家一起成长！！
完
elastic-job 定时任务集成相关推荐
- SpringBoot定时任务 - 集成quartz实现定时任务（单实例和分布式两种方式)
最为常用定时任务框架是Quartz,并且Spring也集成了Quartz的框架,Quartz不仅支持单实例方式还支持分布式方式.本文主要介绍Quartz,基础的Quartz的集成案例本,以及实现基于数 ...
- Elastic Stack容器化部署拓展（Https、AD域集成）并收集Cisco设备的日志信息
前言: 还记得在去年的笔记中提到过EFK(Elasticsearch-Filebeat-Kibana)的部署,但是其中的内容相对简单,也没有提到一些额外的Elastic Stack的特性.链接如下:h ...
- 小米手机v3.exo 合并_Android 3.0预览版SDK已发布，eXo启动了Elastic Beanstalk的IDE
小米手机v3.exo 合并 ECF 3.5速览 斯科特·刘易斯(Scott Lewis)发布了一个博客,提供了对Eclipse Communication Framework(ECF)3.5版所能期望 ...
- SpringBoot + MyBatis-Plus+DM8 +Elasic-job 通用代码框架
目录 1.基础技术框架 2.完整pom.xml 3.初始化脚本 4.项目结构截图: 5.业务模块说明 6.通用模块说明 7.系统初始化配置 8.系统入口 9.配置参数和日志 10.Mapper 配置文 ...
- 这23个非常优秀的SpringBoot/SpringCloud开源项目快收藏起来吧
欢迎大家关注公众号:好码农,了解和学习更多的知识 今天为大家分享23款非常优秀的基于SpringBoot技术体系的开源框架,开箱即用,这些开源项目都具备了很完善的功能,无论是学习还是开发新项目,都会给 ...
- 推荐 7 个牛哄哄 Spring Cloud 实战项目
今日推荐 成都女学霸高考 692 分想当"程序媛",网友:快劝劝孩子 推荐一个非常火爆的电商开源项目! 推荐两个项目! 把一个大型的单个应用程序和服务拆分为数个甚至数十个的支持微服 ...
- 结构化日志：出错时你最想要的好朋友
目录 介绍 什么是日志? Grab中日志的状况 为什么改变? 结构化日志 支持不同格式的多写 开发中类似生产环境的日志 因果顺序 但为什么要结构化记日志? 原文:Structured Logging: ...
- 百度SEO Cloud-Platform(后台管理系统) v3.1.0
简介: Cloud-Platform是后端服务开发的一个十分专业的平台,集成Sentinel,前端采用vue-element-admin组件,Elastic Search自行集成,通过JWT的方式来进 ...
- 基于Spring Cloud微服务化开发平台-Cloud Platform后台管理系统 v3.1.0
介绍: Cloud-Platform是国内首个基于Spring Cloud微服务化开发平台,具有统一授权.认证后台管理系统,其中包含具备用户管理.资源权限管理.网关API 管理等多个模块,支持多业务系 ...
最新文章
- 《富爸爸 辞职创业前的10堂课》文摘
- Spring任务调度
- vue 动态设置页面title
- 【论文解读】通过知识蒸馏促进轻量级深度估计
- 学习Android MediaPlayer
- 跨平台移动应用开发框架AppCan2.0开发经验分享
- spring mvc 提供的几个常用的扩展点
- CodeForces1005D - Polycarp and Div 3
- PyQt5 实现类似海康的设备搜索工具
- sonar pmd\p3c插件源码初步解析
- 2019/第十届蓝桥杯国赛/Java-B
- IBM Spectrum LSF Suites
- .glusterfs_如何在Ubuntu 20.04上使用GlusterFS创建冗余存储池
- 深度学习|费解的tensorflow
- 土豆网王微：做自己生活的导演
- 计算机视觉基础-1——直观简介
- 1一10到时的英文单词_1到10的英文单词
- 【编码实战】2022年还在用jjwt操作jwt？，推荐你使用nimbus-jose-jwt，爽到飞起~
- 保送研究生有什么要求吗？
- win10系统进入英雄联盟后黑屏怎么办