关于java:SpringBoot定时任务-开箱即用分布式任务框架xxljob

除了前文介绍的ElasticJob，xxl-job在很多中小公司有着利用（尽管其代码和设计等品质并不太高，License不够凋谢，有着个人主义色调，然而其具体开箱应用的便捷性和性能绝对欠缺性，这是中小团队采纳的次要起因）；XXL-JOB是一个分布式任务调度平台，其外围设计指标是开发迅速、学习简略、轻量级、易扩大。本文介绍XXL-JOB以及SpringBoot的集成。@pdai

SpringBoot定时工作 - 分布式xxl-job形式
- 常识筹备
  - 分布式工作常识体系
  - 什么是xxl-job
  - xxl-job的架构设计
    - 设计思维
    - 零碎组成
    - 架构图
- 实现案例
  - Bean模式(基于办法)
    - Job的开发环境依赖
    - Job的开发
    - Job的调度配置和执行
  - Bean模式(基于类)
    - Job的开发环境依赖
    - Job的开发
    - Job的调度配置和执行
  - GLUE模式
    - 配置和启动流程
    - GLUE模式还有哪些
  - 更多配置的阐明
- 示例源码

常识筹备

须要对分布式工作的常识体系和xxl-Job有根本的了解。@pdai

什么是xxl-job

XXL-JOB是一个分布式任务调度平台，其外围设计指标是开发迅速、学习简略、轻量级、易扩大。现已凋谢源代码并接入多家公司线上产品线，开箱即用。如下内容来源于xxl-job官网

反对如下个性：

1、简略：反对通过Web页面对工作进行CRUD操作，操作简略，一分钟上手；
2、动静：反对动静批改工作状态、启动/进行工作，以及终止运行中工作，即时失效；
3、调度核心HA（核心式）：调度采纳核心式设计，“调度核心”自研调度组件并反对集群部署，可保障调度核心HA；
4、执行器HA（分布式）：工作分布式执行，工作"执行器"反对集群部署，可保障工作执行HA；
5、注册核心: 执行器会周期性主动注册工作, 调度核心将会主动发现注册的工作并触发执行。同时，也反对手动录入执行器地址；
6、弹性扩容缩容：一旦有新执行器机器上线或者下线，下次调度时将会重新分配工作；
7、触发策略：提供丰盛的工作触发策略，包含：Cron触发、固定距离触发、固定延时触发、API（事件）触发、人工触发、父子工作触发；
8、调度过期策略：调度核心错过调度工夫的弥补解决策略，包含：疏忽、立刻弥补触发一次等；
9、阻塞解决策略：调度过于密集执行器来不及解决时的解决策略，策略包含：单机串行（默认）、抛弃后续调度、笼罩之前调度；
10、工作超时管制：反对自定义工作超时工夫，工作运行超时将会被动中断工作；
11、工作失败重试：反对自定义工作失败重试次数，当工作失败时将会依照预设的失败重试次数被动进行重试；其中分片工作反对分片粒度的失败重试；
12、工作失败告警；默认提供邮件形式失败告警，同时预留扩大接口，可不便的扩大短信、钉钉等告警形式；
13、路由策略：执行器集群部署时提供丰盛的路由策略，包含：第一个、最初一个、轮询、随机、一致性HASH、最不常常应用、最近最久未应用、故障转移、繁忙转移等；
14、分片播送工作：执行器集群部署时，工作路由策略抉择"分片播送"状况下，一次任务调度将会播送触发集群中所有执行器执行一次工作，可依据分片参数开发分片工作；
15、动静分片：分片播送工作以执行器为维度进行分片，反对动静扩容执行器集群从而动静减少分片数量，协同进行业务解决；在进行大数据量业务操作时可显著晋升工作解决能力和速度。
16、故障转移：工作路由策略抉择"故障转移"状况下，如果执行器集群中某一台机器故障，将会主动Failover切换到一台失常的执行器发送调度申请。
17、工作进度监控：反对实时监控工作进度；
18、Rolling实时日志：反对在线查看调度后果，并且反对以Rolling形式实时查看执行器输入的残缺的执行日志；
19、GLUE：提供Web IDE，反对在线开发工作逻辑代码，动静公布，实时编译失效，省略部署上线的过程。反对30个版本的历史版本回溯。
20、脚本工作：反对以GLUE模式开发和运行脚本工作，包含Shell、Python、NodeJS、PHP、PowerShell等类型脚本;
21、命令行工作：原生提供通用命令行工作Handler（Bean工作，"CommandJobHandler"）；业务方只须要提供命令行即可；
22、工作依赖：反对配置子工作依赖，当父工作执行完结且执行胜利后将会被动触发一次子工作的执行, 多个子工作用逗号分隔；
23、一致性：“调度核心”通过DB锁保障集群散布式调度的一致性, 一次任务调度只会触发一次执行；
24、自定义工作参数：反对在线配置调度工作入参，即时失效；
25、调度线程池：调度零碎多线程触发调度运行，确保调度准确执行，不被梗塞；
26、数据加密：调度核心和执行器之间的通信进行数据加密，晋升调度信息安全性；
27、邮件报警：工作失败时反对邮件报警，反对配置多邮件地址群发报警邮件；
28、推送maven地方仓库: 将会把最新稳定版推送到maven地方仓库, 不便用户接入和应用;
29、运行报表：反对实时查看运行数据，如工作数量、调度次数、执行器数量等；以及调度报表，如调度日期分布图，调度胜利分布图等；
30、全异步：任务调度流程全异步化设计实现，如异步调度、异步运行、异步回调等，无效对密集调度进行流量削峰，实践上反对任意时长工作的运行；
31、跨语言：调度核心与执行器提供语言无关的 RESTful API 服务，第三方任意语言可据此对接调度核心或者实现执行器。除此之外，还提供了 “多任务模式”和“httpJobHandler”等其余跨语言计划；
32、国际化：调度核心反对国际化设置，提供中文、英文两种可选语言，默认为中文；
33、容器化：提供官网docker镜像，并实时更新推送dockerhub，进一步实现产品开箱即用；
34、线程池隔离：调度线程池进行隔离拆分，慢工作主动降级进入"Slow"线程池，防止耗尽调度线程，进步零碎稳定性；
35、用户治理：反对在线管理系统用户，存在管理员、普通用户两种角色；
36、权限管制：执行器维度进行权限管制，管理员领有全量权限，普通用户须要调配执行器权限后才容许相干操作；

xxl-job的架构设计

设计思维

将调度行为形象造成“调度核心”公共平台，而平台本身并不承当业务逻辑，“调度核心”负责发动调度申请。

将工作形象成扩散的JobHandler，交由“执行器”对立治理，“执行器”负责接管调度申请并执行对应的JobHandler中业务逻辑。

因而，“调度”和“工作”两局部能够互相解耦，进步零碎整体稳定性和扩展性；

零碎组成

调度模块（调度核心）
1. 负责管理调度信息，依照调度配置收回调度申请，本身不承当业务代码。调度零碎与工作解耦，进步了零碎可用性和稳定性，同时调度零碎性能不再受限于工作模块；
2. 反对可视化、简略且动静的治理调度信息，包含工作新建，更新，删除，GLUE开发和工作报警等，所有上述操作都会实时失效，同时反对监控调度后果以及执行日志，反对执行器Failover。
执行模块（执行器）：
1. 负责接管调度申请并执行工作逻辑。工作模块专一于工作的执行等操作，开发和保护更加简略和高效；
2. 接管“调度核心”的执行申请、终止申请和日志申请等。

架构图

实现案例

次要介绍SpringBoot集成xxl-job的形式：Bean模式（基于办法和基于类）；以及基于在线配置代码/脚本的GLUE模式。

Bean模式(基于办法)

Bean模式工作，反对基于办法的开发方式，每个工作对应一个办法。基于办法开发的工作，底层会生成JobHandler代理，和基于类的形式一样，工作也会以JobHandler的模式存在于执行器工作容器中。

长处：

每个工作只须要开发一个办法，并增加”@XxlJob”注解即可，更加不便、疾速。
反对主动扫描工作并注入到执行器容器。

毛病：要求Spring容器环境；

Job的开发环境依赖

Maven 依赖

<dependency>    <groupId>com.xuxueli</groupId>    <artifactId>xxl-job-core</artifactId>    <version>2.3.1</version></dependency>

application.properties配置

# web portserver.port=8081# no web#spring.main.web-environment=false# log configlogging.config=classpath:logback.xml### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin### xxl-job, access tokenxxl.job.accessToken=default_token### xxl-job executor appnamexxl.job.executor.appname=xxl-job-executor-sample### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is nullxxl.job.executor.address=### xxl-job executor server-infoxxl.job.executor.ip=xxl.job.executor.port=9999### xxl-job executor log-pathxxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler### xxl-job executor log-retention-daysxxl.job.executor.logretentiondays=30

Config配置(PS：这里我是间接拿的xxl-job demo中的配置，理论开发中能够封装一个starter主动注入)

package tech.pdai.springboot.xxljob.config;import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;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.Configuration;/** * xxl-job config * * @author xuxueli 2017-04-28 */@Configurationpublic class XxlJobConfig {    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);    @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() {        logger.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);        // Bean办法模式        // 通过扫描@XxlJob形式注册        // 注册Bean类模式        XxlJobExecutor.registJobHandler("beanClassDemoJobHandler", new BeanClassDemoJob());        return xxlJobSpringExecutor;    }}

Job的开发

开发步骤：

工作开发：在Spring Bean实例中，开发Job办法；
注解配置：为Job办法增加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化办法", destroy = "JobHandler销毁办法")"，注解value值对应的是调度核心新建工作的JobHandler属性的值。
执行日志：须要通过 "XxlJobHelper.log" 打印执行日志；
工作后果：默认工作后果为 "胜利" 状态，不须要被动设置；如有诉求，比方设置工作后果为失败，能够通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置工作后果；

package tech.pdai.springboot.xxljob.job;import java.io.BufferedInputStream;import java.io.BufferedReader;import java.io.DataOutputStream;import java.io.InputStreamReader;import java.net.HttpURLConnection;import java.net.URL;import java.util.Arrays;import com.xxl.job.core.context.XxlJobHelper;import com.xxl.job.core.handler.annotation.XxlJob;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Component;/** * XxlJob开发示例（Bean模式 - 办法） *  */@Slf4j@Componentpublic class BeanMethodDemoJob {    /**     * 1、简略工作示例（Bean模式）     */    @XxlJob("demoJobHandler")    public void demoJobHandler() {        XxlJobHelper.log("demoJobHandler execute...");    }    /**     * 2、分片播送工作     */    @XxlJob("shardingJobHandler")    public void shardingJobHandler() throws Exception {        // logback console日志        log.info("shardingJobHandler execute...");        // 通过xxl记录到DB中的日志        XxlJobHelper.log("shardingJobHandler execute...");        // 分片参数        int shardIndex = XxlJobHelper.getShardIndex();        int shardTotal = XxlJobHelper.getShardTotal();        XxlJobHelper.log("分片参数：以后分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);        // 业务逻辑        for (int i = 0; i < shardTotal; i++) {            if (i==shardIndex) {                XxlJobHelper.log("第 {} 片, 命中分片开始解决", i);            } else {                XxlJobHelper.log("第 {} 片, 疏忽", i);            }        }    }    /**     * 3、命令行工作     */    @XxlJob("commandJobHandler")    public void commandJobHandler() throws Exception {        XxlJobHelper.log("commandJobHandler execute...");        String command = XxlJobHelper.getJobParam();        int exitValue = -1;        BufferedReader bufferedReader = null;        try {            // command process            ProcessBuilder processBuilder = new ProcessBuilder();            processBuilder.command(command);            processBuilder.redirectErrorStream(true);            Process process = processBuilder.start();            //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) {                XxlJobHelper.log(line);            }            // command exit            process.waitFor();            exitValue = process.exitValue();        } catch (Exception e) {            XxlJobHelper.log(e);        } finally {            if (bufferedReader!=null) {                bufferedReader.close();            }        }        if (exitValue==0) {            // default success        } else {            XxlJobHelper.handleFail("command exit value(" + exitValue + ") is failed");        }    }    /**     * 4、跨平台Http工作     * 参数示例：     * "url: http://www.baidu.com\n" +     * "method: get\n" +     * "data: content\n";     */    @XxlJob("httpJobHandler")    public void httpJobHandler() throws Exception {        XxlJobHelper.log("httpJobHandler execute...");        // param parse        String param = XxlJobHelper.getJobParam();        if (param==null || param.trim().length()==0) {            XxlJobHelper.log("param[" + param + "] invalid.");            XxlJobHelper.handleFail();            return;        }        String[] httpParams = param.split("\n");        String url = null;        String method = null;        String data = null;        for (String httpParam : httpParams) {            if (httpParam.startsWith("url:")) {                url = httpParam.substring(httpParam.indexOf("url:") + 4).trim();            }            if (httpParam.startsWith("method:")) {                method = httpParam.substring(httpParam.indexOf("method:") + 7).trim().toUpperCase();            }            if (httpParam.startsWith("data:")) {                data = httpParam.substring(httpParam.indexOf("data:") + 5).trim();            }        }        // param valid        if (url==null || url.trim().length()==0) {            XxlJobHelper.log("url[" + url + "] invalid.");            XxlJobHelper.handleFail();            return;        }        if (method==null || !Arrays.asList("GET", "POST").contains(method)) {            XxlJobHelper.log("method[" + method + "] invalid.");            XxlJobHelper.handleFail();            return;        }        boolean isPostMethod = method.equals("POST");        // request        HttpURLConnection connection = null;        BufferedReader bufferedReader = null;        try {            // connection            URL realUrl = new URL(url);            connection = (HttpURLConnection) realUrl.openConnection();            // connection setting            connection.setRequestMethod(method);            connection.setDoOutput(isPostMethod);            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();            // data            if (isPostMethod && data!=null && data.trim().length() > 0) {                DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());                dataOutputStream.write(data.getBytes("UTF-8"));                dataOutputStream.flush();                dataOutputStream.close();            }            // 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();            XxlJobHelper.log(responseMsg);            return;        } catch (Exception e) {            XxlJobHelper.log(e);            XxlJobHelper.handleFail();            return;        } finally {            try {                if (bufferedReader!=null) {                    bufferedReader.close();                }                if (connection!=null) {                    connection.disconnect();                }            } catch (Exception e2) {                XxlJobHelper.log(e2);            }        }    }    /**     * 5、生命周期工作示例：工作初始化与销毁时，反对自定义相干逻辑；     */    @XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy")    public void demoJobHandler2() throws Exception {        XxlJobHelper.log("demoJobHandler2, execute...");    }    public void init() {        log.info("init");    }    public void destroy() {        log.info("destroy");    }}

(@pdai: 从设计的角度，xxl-job能够对上述不同类型进行细分)

Job的调度配置和执行

新增Job, 并把上述的@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化办法", destroy = "JobHandler销毁办法")中 自定义jobhandler名称 填写到JobHandler中。

其它配置如下：

能够抉择操作中执行一次工作，或者启动（依照Cron执行)

能够查看执行的记录

进一步能够看每个执行记录的执行日志

Bean模式(基于类)

Bean模式工作，反对基于类的开发方式，每个工作对应一个Java类。

长处：不限度我的项目环境，兼容性好。即便是无框架我的项目，如main办法间接启动的我的项目也能够提供反对，能够参考示例我的项目 “xxl-job-executor-sample-frameless”；

毛病：

每个工作须要占用一个Java类，造成类的节约；
不反对主动扫描工作并注入到执行器容器，须要手动注入。

Job的开发环境依赖

同Bean模式(基于办法)

Job的开发

开发步骤：

执行器我的项目中，开发Job类：
- 开发一个继承自"com.xxl.job.core.handler.IJobHandler"的JobHandler类，实现其中工作办法。
- 手动通过如下形式注入到执行器容器。
注册jobHandler
- XxlJobExecutor.registJobHandler("xxxxxJobHandler", new xxxxxJobHandler());

Job开发

package tech.pdai.springboot.xxljob.job;import com.xxl.job.core.handler.IJobHandler;import lombok.extern.slf4j.Slf4j;/** * @author pdai */@Slf4jpublic class BeanClassDemoJob extends IJobHandler {    @Override    public void execute() throws Exception {        log.info("BeanClassDemoJob, execute...");    }}

注册jobHandler(@pdai: 这里xxl-job设计的不好，是能够通过IJobHandler来主动注册的)

XxlJobExecutor.registJobHandler("beanClassDemoJobHandler", new BeanClassDemoJob());

启动SpringBoot利用, 能够发现注册的

...20:34:15.385 logback [main] INFO  c.x.job.core.executor.XxlJobExecutor - >>>>>>>>>>> xxl-job register jobhandler success, name:beanClassDemoJobHandler, jobHandler:tech.pdai.springboot.xxljob.job.BeanClassDemoJob@640ab13c...

Job的调度配置和执行

同Bean模式(基于办法)

在调度器中增加执行后，后盾执行的日志如下：

20:41:00.021 logback [xxl-job, EmbedServer bizThreadPool-1023773196] INFO  c.x.job.core.executor.XxlJobExecutor - >>>>>>>>>>> xxl-job regist JobThread success, jobId:5, handler:tech.pdai.springboot.xxljob.job.BeanClassDemoJob@640ab13c20:41:00.022 logback [xxl-job, JobThread-5-1654681260021] INFO  t.p.s.xxljob.job.BeanClassDemoJob - BeanClassDemoJob, execute...

GLUE模式

工作以源码形式保护在调度核心，反对通过Web IDE在线更新，实时编译和失效，因而不须要指定JobHandler。

配置和启动流程

开发流程如下：

创立GLUE类型的Job（这里以Java为例）

选中指定工作，点击该工作右侧“GLUE”按钮，将会返回GLUE工作的Web IDE界面，在该界面反对对工作代码进行开发（也能够在IDE中开发实现后，复制粘贴到编辑中）。

版本回溯性能（反对30个版本的版本回溯）：在GLUE工作的Web IDE界面，抉择右上角下拉框“版本回溯”，会列出该GLUE的更新历史，抉择相应版本即可显示该版本代码，保留后GLUE代码即回退到对应的历史版本；

执行后的记录如下

GLUE模式还有哪些

xxl-job一共反对如下几种GLUE模式：

GLUE模式(Java)：工作以源码形式保护在调度核心；该模式的工作实际上是一段继承自IJobHandler的Java类代码并 "groovy" 源码形式保护，它在执行器我的项目中运行，可应用@Resource/@Autowire注入执行器里中的其余服务；
GLUE模式(Shell)：工作以源码形式保护在调度核心；该模式的工作实际上是一段 "shell" 脚本；
GLUE模式(Python)：工作以源码形式保护在调度核心；该模式的工作实际上是一段 "python" 脚本；
GLUE模式(PHP)：工作以源码形式保护在调度核心；该模式的工作实际上是一段 "php" 脚本；
GLUE模式(NodeJS)：工作以源码形式保护在调度核心；该模式的工作实际上是一段 "nodejs" 脚本；
GLUE模式(PowerShell)：工作以源码形式保护在调度核心；该模式的工作实际上是一段 "PowerShell" 脚本；

示例源码

https://github.com/realpdai/t...

常识筹备

什么是xxl-job

xxl-job的架构设计

设计思维

零碎组成

架构图

实现案例

Bean模式(基于办法)

Job的开发环境依赖

Job的开发

Job的调度配置和执行

Bean模式(基于类)

Job的开发环境依赖

Job的开发

Job的调度配置和执行

GLUE模式

配置和启动流程

GLUE模式还有哪些

更多配置的阐明

示例源码

更多内容