quartz的不足
Quartz作为开源作业调度中的佼佼者,是作业调度的首选。但是集群环境中Quartz采用API的方式对任务进行管理,从而可以避免上述问题,但是同样存在以下问题:
- 问题一:调用API的的方式操作任务,不人性化;
- 问题二:需要持久化业务QuartzJobBean到底层数据表中,系统侵入性相当严重。
- 问题三:调度逻辑和QuartzJobBean耦合在同一个项目中,这将导致一个问题,在调度任务数量逐渐增多,同时调度任务逻辑逐渐加重的情况下,此时调度系统的性能将大大受限于业务;
- 问题四:quartz底层以“抢占式”获取DB锁并由抢占成功节点负责运行任务,会导致节点负载悬殊非常大;而XXL-JOB通过执行器实现“协同分配式”运行任务,充分发挥集群优势,负载各节点均衡。
XXL-JOB弥补了quartz的上述不足之处。
xxl-job官方文档:https://www.xuxueli.com/xxl-job/
XXL-JOB是一个分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线,开箱即用。
-
先部署调度中心,克隆地址:https://github.com/xuxueli/xxl-job,本文用的是2.2.0-SNAPSHOT,需要下载对应版本,步骤如下
- 克隆完代码后,进入doc/db目录,将sql复制到MySQL执行一遍创建数据库。
- 进入xxl-job-admin目录,修改application.properties文件,将数据源的配置信息修改为自己刚刚创建的数据库。
- 启动xxl-job-admin项目,访问http://localhost:8080/xxl-job-admin,用户名:admin,密码:123456,到此调度中心部署完毕。
运行调度器,登录后效果:
- 部署执行器,克隆的代码里面有官方的示例项目,可以直接运行,也可以集成到自己的项目中,下面集成到自己的项目,在pom.xml加入依赖,如下
<!-- 任务调度xxl-job --> <dependency> <groupId>com.xuxueli</groupId> <artifactId>xxl-job-core</artifactId> <version>2.2.0</version> </dependency>
- 创建XxlJobConfig配置类
package com.xj.demo.config; import com.xxl.job.core.executor.impl.XxlJobSpringExecutor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Slf4j @Configuration public class XxlJobConfig { @Value("${xxl.job.admin.addresses}") private String adminAddresses; @Value("${xxl.job.accessToken}") private String accessToken; @Value("${xxl.job.executor.appname}") private String appname; @Value("${xxl.job.executor.address}") private String address; @Value("${xxl.job.executor.ip}") private String ip; @Value("${xxl.job.executor.port}") private int port; @Value("${xxl.job.executor.logpath}") private String logPath; @Value("${xxl.job.executor.logretentiondays}") private int logRetentionDays; @Bean public XxlJobSpringExecutor xxlJobExecutor() { log.info(">>>>>>>>>>> xxl-job config init."); XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor(); xxlJobSpringExecutor.setAdminAddresses(adminAddresses); xxlJobSpringExecutor.setAppname(appname); xxlJobSpringExecutor.setAddress(address); xxlJobSpringExecutor.setIp(ip); xxlJobSpringExecutor.setPort(port); xxlJobSpringExecutor.setAccessToken(accessToken); xxlJobSpringExecutor.setLogPath(logPath); xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays); return xxlJobSpringExecutor; } /** * 针对多网卡、容器内部署等情况,可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP; * * 1、引入依赖: * <dependency> * <groupId>org.springframework.cloud</groupId> * <artifactId>spring-cloud-commons</artifactId> * <version>${version}</version> * </dependency> * * 2、配置文件,或者容器启动变量 * spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.' * * 3、获取IP * String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress(); */ }
- 创建XxlJobService类,重点关注@XxlJob,后面要用到
package com.xj.demo.service; import com.xj.demo.annotation.JwtIgnore; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.handler.IJobHandler; import com.xxl.job.core.handler.annotation.XxlJob; import com.xxl.job.core.log.XxlJobLogger; import com.xxl.job.core.util.ShardingUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.io.BufferedInputStream; import java.io.BufferedReader; import java.io.InputStreamReader; import java.net.HttpURLConnection; import java.net.URL; import java.util.concurrent.TimeUnit; /** * XxlJob开发示例(Bean模式) * * 开发步骤: * 1、在Spring Bean实例中,开发Job方法,方式格式要求为 "public ReturnT<String> execute(String param)" * 2、为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")",注解value值对应的是调度中心新建任务的JobHandler属性的值。 * 3、执行日志:需要通过 "XxlJobLogger.log" 打印执行日志; * * @author xuxueli 2019-12-11 21:52:51 */ @Slf4j @Component public class XxlJobService { /** * 1、简单任务示例(Bean模式) */ @XxlJob("demoJobHandler") @JwtIgnore public ReturnT<String> demoJobHandler(String param) throws Exception { XxlJobLogger.log("XXL-JOB, Hello World."); log.info("执行器执行成功"); for (int i = 0; i < 5; i++) { XxlJobLogger.log("beat at:" + i); TimeUnit.SECONDS.sleep(2); } return ReturnT.SUCCESS; } /** * 2、分片广播任务 */ @XxlJob("shardingJobHandler") public ReturnT<String> shardingJobHandler(String param) throws Exception { // 分片参数 ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo(); XxlJobLogger.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal()); // 业务逻辑 for (int i = 0; i < shardingVO.getTotal(); i++) { if (i == shardingVO.getIndex()) { XxlJobLogger.log("第 {} 片, 命中分片开始处理", i); } else { XxlJobLogger.log("第 {} 片, 忽略", i); } } return ReturnT.SUCCESS; } /** * 3、命令行任务 */ @XxlJob("commandJobHandler") public ReturnT<String> commandJobHandler(String param) throws Exception { String command = param; int exitValue = -1; BufferedReader bufferedReader = null; try { // command process Process process = Runtime.getRuntime().exec(command); BufferedInputStream bufferedInputStream = new BufferedInputStream(process.getInputStream()); bufferedReader = new BufferedReader(new InputStreamReader(bufferedInputStream)); // command log String line; while ((line = bufferedReader.readLine()) != null) { XxlJobLogger.log(line); } // command exit process.waitFor(); exitValue = process.exitValue(); } catch (Exception e) { XxlJobLogger.log(e); } finally { if (bufferedReader != null) { bufferedReader.close(); } } if (exitValue == 0) { return IJobHandler.SUCCESS; } else { return new ReturnT<String>(IJobHandler.FAIL.getCode(), "command exit value("+exitValue+") is failed"); } } /** * 4、跨平台Http任务 */ @XxlJob("httpJobHandler") public ReturnT<String> httpJobHandler(String param) throws Exception { // request HttpURLConnection connection = null; BufferedReader bufferedReader = null; try { // connection URL realUrl = new URL(param); connection = (HttpURLConnection) realUrl.openConnection(); // connection setting connection.setRequestMethod("GET"); connection.setDoOutput(true); connection.setDoInput(true); connection.setUseCaches(false); connection.setReadTimeout(5 * 1000); connection.setConnectTimeout(3 * 1000); connection.setRequestProperty("connection", "Keep-Alive"); connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8"); connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8"); // do connection connection.connect(); //Map<String, List<String>> map = connection.getHeaderFields(); // valid StatusCode int statusCode = connection.getResponseCode(); if (statusCode != 200) { throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid."); } // result bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8")); StringBuilder result = new StringBuilder(); String line; while ((line = bufferedReader.readLine()) != null) { result.append(line); } String responseMsg = result.toString(); XxlJobLogger.log(responseMsg); return ReturnT.SUCCESS; } catch (Exception e) { XxlJobLogger.log(e); return ReturnT.FAIL; } finally { try { if (bufferedReader != null) { bufferedReader.close(); } if (connection != null) { connection.disconnect(); } } catch (Exception e2) { XxlJobLogger.log(e2); } } } /** * 5、生命周期任务示例:任务初始化与销毁时,支持自定义相关逻辑; */ @XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy") public ReturnT<String> demoJobHandler2(String param) throws Exception { XxlJobLogger.log("XXL-JOB, Hello World."); return ReturnT.SUCCESS; } public void init(){ log.info("init"); } public void destroy(){ log.info("destory"); } }
- 在配置文件中,添加如下配置
spring: application: name: xxljob-demo-server server: port: 8090 # xxl-job配置 xxl: job: admin: # 调度中心部署跟地址 [选填]:如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调";为空则关闭自动注册; addresses: http://localhost:8080/xxl-job-admin executor: # 执行器注册 [选填]:优先使用该配置作为注册地址,为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。 address: # 执行器AppName [选填]:执行器心跳注册分组依据;为空则关闭自动注册 appname: demo-app # 执行器IP [选填]:默认为空表示自动获取IP,多网卡时可手动设置指定IP,该IP不会绑定Host仅作为通讯实用;地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"; ip: # 执行器端口号 [选填]:小于等于0则自动获取;默认端口为9999,单机部署多个执行器时,注意要配置不同执行器端口; port: 9999 # 执行器运行日志文件存储磁盘路径 [选填] :需要对该路径拥有读写权限;为空则使用默认路径; logpath: /Users/luoyu/Documents/log/xxl-job/jobhandler # 执行器日志文件保存天数 [选填] : 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能; logretentiondays: 15 # 执行器通讯TOKEN [选填]:非空时启用; accessToken:
注意:
- 重点关注appname,后面要用到
- 需要调度中心跟执行器进行认证的话,把两者的accessToken配置一样的即可
- 配置好的完整目录结构如下:
- 使用流程如下:
- 启动调度中心项目,创建执行器,如图
- 在任务管理,选择上图创建的执行器
![](https://img2020.cnblogs.com/blog/300118/202012/300118-20201203173218870-618570714.png)
xxl-job 架构图
转载:https://www.jianshu.com/p/20ac45a76fde