代码已提交至Github,有兴趣的同学可以下载来看看:https://github.com/ylw-github/SpringBoot-XXLJOB-Demo
本文目录结构:
l________1.3 使用ScheduledExecutorService
l____________3.2.1 部署 xxl-job-admin 注册中心
l____________3.2.3 在调度中心中配置执行器
1. 实现任务调度的方案
任务调度指的是指定时间去执行指定的任务,那么Java有哪几种实现任务调度的方式呢?直接贴上代码来理解:
1.1 使用Thread
public class ThreadDemo { static long count = 0; public static void main(String[] args) { Runnable runnable = new Runnable() { @Override public void run() { while (true) { try { Thread.sleep(1000); count++; System.out.println(count); } catch (Exception e) { // TODO: handle exception } } } }; Thread thread = new Thread(runnable); thread.start(); } }
1.2 使用TimerTask
/** * 使用TimerTask类实现定时任务 */ public class TimerTaskDemo { static long count = 0; public static void main(String[] args) { TimerTask timerTask = new TimerTask() { @Override public void run() { count++; System.out.println(count); } }; Timer timer = new Timer(); // 天数 long delay = 0; // 秒数 long period = 1000; timer.scheduleAtFixedRate(timerTask, delay, period); } }
1.3 使用ScheduledExecutorService
ScheduledExecutorService是从JavaSE5的java.util.concurrent
里,做为并发工具类被引进的,这是最理想的定时任务实现方式。
public class ScheduledExecutorServiceDemo { public static void main(String[] args) { Runnable runnable = new Runnable() { public void run() { // task to run goes here System.out.println("Hello !!"); } }; ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); // 第二个参数为首次执行的延时时间,第三个参数为定时执行的间隔时间 service.scheduleAtFixedRate(runnable, 1, 1, TimeUnit.SECONDS); } }
1.4 使用Quartz
操作步骤:
1.引入maven依赖
<dependencies> <!-- quartz --> <dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId> <version>2.2.1</version> </dependency> <dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz-jobs</artifactId> <version>2.2.1</version> </dependency> </dependencies>
2.任务调度类
public class MyJob implements Job { public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println("quartz MyJob date:" + new Date().getTime()); } }
3.启动类
//1.创建Scheduler的工厂 SchedulerFactory sf = new StdSchedulerFactory(); //2.从工厂中获取调度器实例 Scheduler scheduler = sf.getScheduler(); //3.创建JobDetail JobDetail jb = JobBuilder.newJob(MyJob.class) .withDescription("this is a ram job") //job的描述 .withIdentity("ramJob", "ramGroup") //job 的name和group .build(); //任务运行的时间,SimpleSchedle类型触发器有效 long time= System.currentTimeMillis() + 3*1000L; //3秒后启动任务 Date statTime = new Date(time); //4.创建Trigger //使用SimpleScheduleBuilder或者CronScheduleBuilder Trigger t = TriggerBuilder.newTrigger() .withDescription("") .withIdentity("ramTrigger", "ramTriggerGroup") //.withSchedule(SimpleScheduleBuilder.simpleSchedule()) .startAt(statTime) //默认当前时间启动 .withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?")) //两秒执行一次 .build(); //5.注册任务和定时器 scheduler.scheduleJob(jb, t); //6.启动 调度器 scheduler.start();
4.Quartz表达式可以参考:
2. 定时任务在分布式下的解决方案
分布式集群的情况下,怎么保证定时任务不被重复执行?
有如下几个方案:
- 使用zookeeper实现分布式锁 (缺点:需要创建临时节点、和事件通知不易于扩展)。
- 使用配置文件做一个开关 (缺点:发布后,需要重启)。
- 数据库唯一约束( 缺点:效率低)。
推荐
:使用分布式任务调度平台XXLJOB、Elastric-Job、TBSchedule。
3. XXLJOB
3.1 XXLJOB介绍
- 简单:支持通过Web页面对任务进行CRUD操作,操作简单,一分钟上手;
- 动态:支持动态修改任务状态、暂停/恢复任务,以及终止运行中任务,即时生效;
- 调度中心HA(中心式) :调度采用中心式设计,“调度中心”基于集群Quartz实现,可保证调度中心HA;
- 执行器HA(分布式):任务分布式执行,任务"执行器"支持集群部署,可保证任务执行HA;
- 任务Failover:执行器集群部署时,任务路由策略选择"故障转移"情况下调度失败时将会平滑切换执行器进行Failover;
- 一致性:“调度中心”通过DB锁保证集群分布式调度的一致性, 一次任务调度只会触发一次执行;
- 自定义任务参数:支持在线配置调度任务入参,即时生效;
- 调度线程池:调度系统多线程触发调度运行,确保调度精确执行,不被堵塞;
- 弹性扩容缩容:一旦有新执行器机器上线或者下线,下次调度时将会重新分配任务
- 邮件报警:任务失败时支持邮件报警,支持配置多邮件地址群发报警邮件;
- 状态监控:支持实时监控任务进度;
- Rolling执行日志:支持在线查看调度结果,并且支持以Rolling方式实时查看执行器输出的完整的执行日志;
- GLUE:提供Web IDE,支持在线开发任务逻辑代码,动态发布,实时编译生效,省略部署上线的过程。支持30个版本的历史版本回溯。
- 数据加密:调度中心和执行器之间的通讯进行数据加密,提升调度信息安全性;
- 任务依赖:支持配置子任务依赖,当父任务执行结束且执行成功后将会主动触发一次子任务的执行, 多个子任务用逗号分隔;
- 推送maven中央仓库: 将会把最新稳定版推送到maven中央仓库, 方便用户接入和使用;
- 任务注册: 执行器会周期性自动注册任务, 调度中心将会自动发现注册的任务并触发执行。同时,也支持手动录入执行器地址;
- 路由策略:执行器集群部署时提供丰富的路由策略,包括:第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等;
- 运行报表:支持实时查看运行数据,如任务数量、调度次数、执行器数量等;以及调度报表,如调度日期分布图,调度成功分布图等;
- 脚本任务:支持以GLUE模式开发和运行脚本任务,包括Shell、Python等类型脚本;
- 阻塞处理策略:调度过于密集执行器来不及处理时的处理策略,策略包括:单机串行(默认)、丢弃后续调度、覆盖之前调度;
- 失败处理策略;调度失败时的处理策略,策略包括:失败告警(默认)、失败重试;
- 分片广播任务:执行器集群部署时,任务路由策略选择"分片广播"情况下,一次任务调度将会广播触发对应集群中所有执行器执行一次任务,同时传递分片参数;可根据分片参数开发分片任务;
- 动态分片:分片广播任务以执行器为维度进行分片,支持动态扩容执行器集群从而动态增加分片数量,协同进行业务处理;在进行大数据量业务操作时可显著提升任务处理能力和速度。
- 事件触发:除了"Cron方式"和"任务依赖方式"触发任务执行之外,支持基于事件的触发任务方式。调度中心提供触发任务单次执行的API服务,可根据业务事件灵活触发。
源码Github地址:https://github.com/xuxueli/xxl-job
文档地址https://www.xuxueli.com/xxl-job/#/:
3.2 XXLJOB集成
集成步骤:
1. 部署: xxl-job-admin 作为注册中心 2.创建执行器项目(具体调度地址) 可以支持集群 ------ 配置文件需要填写xxl-job注册中心地址 ------ 每个具体执行job服务器需要创建一个netty连接端口号 ------ 需要执行job的任务类,集成IJobHandler抽象类注册到job容器中 ------ Execute方法中编写具体job任务 3. 在调度中心中配置执行器 4. 启动任务
3.2.1 部署 xxl-job-admin 注册中心
1.使用Git Clone源码到IDEA:https://github.com/xuxueli/xxl-job.git
2.执行数据库脚本:/xxl-job/doc/db/tables_xxl_job.sql
,可以看到产生了8张表。
3.配置调度中心
调度中心就是源码中的 xxl-job-admin 工程,配置文件地址:/xxl-job/xxl-job-admin/src/main/resources/application.properties
### web server.port=8080 server.context-path=/xxl-job-admin ### actuator management.context-path=/actuator management.health.mail.enabled=false ### resources spring.mvc.static-path-pattern=/static/** spring.resources.static-locations=classpath:/static/ ### freemarker spring.freemarker.templateLoaderPath=classpath:/templates/ spring.freemarker.suffix=.ftl spring.freemarker.charset=UTF-8 spring.freemarker.request-context-attribute=request spring.freemarker.settings.number_format=0.########## ### mybatis mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml ### xxl-job, datasource spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?Unicode=true&characterEncoding=UTF-8 spring.datasource.username=root spring.datasource.password=123456 spring.datasource.driver-class-name=com.mysql.jdbc.Driver spring.datasource.type=org.apache.tomcat.jdbc.pool.DataSource spring.datasource.tomcat.max-wait=10000 spring.datasource.tomcat.max-active=30 spring.datasource.tomcat.test-on-borrow=true spring.datasource.tomcat.validation-query=SELECT 1 spring.datasource.tomcat.validation-interval=30000 ### xxl-job email spring.mail.host=smtp.qq.com spring.mail.port=25 spring.mail.username=xxx@qq.com spring.mail.password=xxx spring.mail.properties.mail.smtp.auth=true spring.mail.properties.mail.smtp.starttls.enable=true spring.mail.properties.mail.smtp.starttls.required=true spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory ### xxl-job, access token xxl.job.accessToken= ### xxl-job, i18n (default empty as chinese, "en" as english) xxl.job.i18n= ## xxl-job, triggerpool max size xxl.job.triggerpool.fast.max=200 xxl.job.triggerpool.slow.max=100 ### xxl-job, log retention days xxl.job.logretentiondays=30
4.启动调度中心,启动类在:/xxl-job/xxl-job-admin/src/main/java/com/xxl/job/admin/XxlJobAdminApplication.java
提示日志文件找不到错误:
其实是没有文件目录造成的,直接修改logback文件(目录:/xxl-job/xxl-job-admin/src/main/resources/logback.xml
),把日志目录改为当前:
<property name="log.path" value="./xxl-job-admin.log"/>
5.进入调度中心http://localhost:8080/xxl-job-admin/toLogin,登录账号admin,密码123456。
3.2.2 创建执行器项目
1.添加Maven依赖(注意版本号,太低的好像有问题):
<!-- xxl-rpc-core --> <dependency> <groupId>com.xuxueli</groupId> <artifactId>xxl-job-core</artifactId> <version>2.1.1</version> </dependency>
2.配置执行器(application.yml):
server: #项目端口号 port: 8081 logging: #日志文件 config: classpath:logback.xml xxl: job: admin: #调度中心部署跟地址:如调度中心集群部署存在多个地址则用逗号分隔。 #执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"。 addresses: http://127.0.0.1:8080/xxl-job-admin accessToken: #分别配置执行器的名称、ip地址、端口号 #注意:如果配置多个执行器时,防止端口冲突 executor: appname: executorDemo1 ip: 127.0.0.1 port: 9999 #执行器运行日志文件存储的磁盘位置,需要对该路径拥有读写权限 logpath: /data/applogs/xxl-job/jobhandler #执行器Log文件定期清理功能,指定日志保存天数,日志文件过期自动删除。限制至少保持3天,否则功能不生效; #-1表示永不删除 logretentiondays: -1
3.载入配置文件(新增XxlJobConfig文件):
package com.ylw.job.executor.config; import com.xxl.job.core.executor.XxlJobExecutor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; @Configuration @ComponentScan(basePackages = "com.ylw.job.executor.jobhandler") public class XxlJobConfig { private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class); @Value("${xxl.job.admin.addresses}") private String adminAddresses; @Value("${xxl.job.executor.appname}") private String appName; @Value("${xxl.job.executor.ip}") private String ip; @Value("${xxl.job.executor.port}") private int port; @Value("${xxl.job.accessToken}") private String accessToken; @Value("${xxl.job.executor.logpath}") private String logPath; @Value("${xxl.job.executor.logretentiondays}") private int logRetentionDays; @Bean(initMethod = "start", destroyMethod = "destroy") public XxlJobExecutor xxlJobExecutor() { logger.info(">>>>>>>>>>> xxl-job config init."); XxlJobExecutor xxlJobExecutor = new XxlJobExecutor(); xxlJobExecutor.setAdminAddresses(adminAddresses); xxlJobExecutor.setAppName(appName); xxlJobExecutor.setIp(ip); xxlJobExecutor.setPort(port); xxlJobExecutor.setAccessToken(accessToken); xxlJobExecutor.setLogPath(logPath); xxlJobExecutor.setLogRetentionDays(logRetentionDays); return xxlJobExecutor; } }
4.创建任务JobHandler
@JobHandler("everySecondsHandler") @Component public class EverySecondsHandler extends IJobHandler { private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class); @Value("${server.port}") private String serverPort; @Override public ReturnT<String> execute(String param) throws Exception { logger.info("######端口号:serverPort" + serverPort + "###定时Job开始执行啦!!!!######"); return SUCCESS; } }
3.2.3 在调度中心中配置执行器
1.新增执行器:
- AppName:是每个执行器集群的唯一标示AppName, 执行器会周期性以AppName为对象进行自动注册。可通过该配置自动发现注册成功的执行器, 供任务调度时使用;
- 名称:执行器的名称, 因为AppName限制字母数字等组成,可读性不强, 名称为了提高执行器的可读性;
- 排序: 执行器的排序, 系统中需要执行器的地方,如任务新增, 将会按照该排序读取可用的执行器列表;
- 注册方式:调度中心获取执行器地址的方式,
- 自动注册:执行器自动进行执行器注册,调度中心通过底层注册表可以动态发现执行器机器地址;
- 手动录入:人工手动录入执行器的地址信息,多地址逗号分隔,供调度中心使用;
- 机器地址:"注册方式"为"手动录入"时有效,支持人工维护执行器的地址信息;
2.新增任务
3.2.4 测试
1.启动SpringBoot项目
2.控制台启动任务:
可以看到日志打印:
3.3 XXL-JOB集群
详情可以参考博客:《XXL-Job高可用集群搭建》
- 调度中心支持集群部署,提升调度系统容灾和可用性。
- 调度中心集群部署时,几点要求和建议:
- DB配置保持一致;
- 登陆账号配置保持一致;
- 群机器时钟保持一致(单机集群忽视);
建议:推荐通过nginx为调度中心集群做负载均衡,分配域名。调度中心访问、执行器回调配置、调用API服务等操作均通过该域名进行。
Nginx配置信息:
upstream backServer{ server 127.0.0.1:8080 weight=1; server 127.0.0.1:8081 weight=1; } server { listen 80; server_name www.xxx.com; #charset koi8-r; #access_log logs/host.access.log main; location / { proxy_pass http://backServer; index index.html index.htm; } error_page 500 502 503 504 /50x.html; location = /50x.html { root html; } }
总结