1,什么是Elastic-job

Elastic-job 是一个分布式调度解决方案。

Elastic-Job由2个独立的子项目组成：Elastic-Job-Lite和Elastic-Job-Cloud。

Elastic-Job-Lite是一个无中心的解决方案，使用轻量级jar来协调分布式作业。
Elastic-Job-Lite和Elastic-Job-Cloud提供统一的API。开发人员只需要一次代码，然后决定根据需要部署Lite或Cloud。

2，特征

分布式计划作业坐标；

支持弹性缩放和缩小；

故障转移；

Web控制台；

Spring的整合和命名空间的支持；

分布式环境不稳定时自我诊断和恢复；

支持并行调度；
2，使用

第一步：pox.xml文件中引入
      <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-core</artifactId>
        <version>2.1.5</version>
      </dependency>
      <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-spring</artifactId>
        <version>2.1.5</version>
      </dependency>

第二步：Java代码的编写
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;

/**
  @Auth :  zhangna
*/
public class MyElasticJob implements SimpleJob {
     @Override
     public void execute(ShardingContext shardingContext) {
     
           you can do something........
           
     }
}

第三步：作业的配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xmlns:reg="http://www.dangdang.com/schema/ddframe/reg"
   xmlns:job="http://www.dangdang.com/schema/ddframe/job"
   xsi:schemaLocation="http://www.springframework.org/schema/beans
                    http://www.springframework.org/schema/beans/spring-beans.xsd
                    http://www.dangdang.com/schema/ddframe/reg
                    http://www.dangdang.com/schema/ddframe/reg/reg.xsd
                    http://www.dangdang.com/schema/ddframe/job
                    http://www.dangdang.com/schema/ddframe/job/job.xsd
                    ">
<!--配置注册中心 -->
<reg:zookeeper id="regCenter" server-lists="127.0.0.1:2181" namespace="dd-job" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" />

<!-- 配置任务-->
<job:simple id="myElasticJob" class="com.zhangna.job.myElasticJob" registry-center-ref="regCenter" cron="0 0/1 * * * ?" sharding-total-count="1" overwrite="true" />
</beans> 

前言
任务调度是java项目中常用的一种组件,可以指定任务在何时进行触发,最熟悉的是spring框架里面的quartz,较流行的有一些分布式调度组件,比如elastic-job/azkaban,都是基于quartz二次开发的,今天介绍一款分布式的任务调度器xxl-job。
项目介绍
xxl-job是一款极容易学习上手的轻量级开源分布式调度框架,分为管理端和执行端两块,管理端负责配置任务信息以及查看任务执行日志,执行端只需要配置与管理端的连接信息就可以进行具体的任务逻辑开发了,目前版本还在持续迭代中,使用简单,功能强大,具体功能特性可以看下官方介绍。废话不多说,直接进入实战吧。
实战
1.服务端部署
从https://github.com/xuxueli/xxl-job下载项目,用mysql客户端工具Navicat执行项目根目录下doc/db/table_xxl_job.sql文件,库名自己可以自行修改,一共8张表,如下:
创建一个新的spring boot项目,将下载的xxl-job-admin目录下的文件以及pom.xml文件都拷贝到新建的项目中(如果不想新建项目可以直接用下载下来的项目进行修改部署),修改application.properties中的数据库连接信息。
小编是自己创建的新项目,需要手动改了pom.xml依赖xxl-job-core的版本为2.2.0
修改logback.xml中的日志输出路径。
好了,以上3步曲就搞定整个服务端配置了,启动项目,并访问http://localhost:8080/xxl-job-admin/ ,默认管理员账号admin/123456进行登录。
这交互,可以啊,是不是很带感。
2.执行端配置
创建一个新的module,跟服务端一样,也需要修改下logback.xml以及在pom.xml添加xxl-job-core的依赖。
为了模拟分布式效果,小编创建了2个配置文件来区分2个执行服务。
application-9998.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=### xxl-job executor appnamexxl.job.executor.appname=my-job-executor### 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=9998### xxl-job executor log-pathxxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler### xxl-job executor log-retention-daysxxl.job.executor.logretentiondays=30
application-9999.properties
# web portserver.port=8082# 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=### xxl-job executor appnamexxl.job.executor.appname=my-job-executor### 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
细心的童鞋会发现只有server.port和xxl.job.executor.port不同,执行器服务跟spring boot一样,自带内嵌tomcat,也会暴露一个端口注册到服务端,进行高可用负载。
创建一个java config类,定义一个使用配置的XxlJobSpringExecutor执行类,如下
@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);        return xxlJobSpringExecutor;    }
配置2个启动配置,分别启动,效果如下:
完美启动2个服务,看下服务端平台是不是有这两台执行服务的注册信息。
注意:为了演示,事先创建了一个执行器,AppName一定要与配置文件中xxl.job.executor.appname一致。
3.任务开发
3.1 基于方法注解任务
话不多说,直接上代码吧,毕竟代码是程序员最好的交流方式。
    /**     * 1、注解任务     */    @XxlJob(value = "myJobAnnotationHandler",init = "init", destroy = "destroy")    public ReturnT myJobAnnotationHandler(String param) throws Exception {        XxlJobLogger.log("XXL-JOB-ANNOTATION, myJobAnnotationHandler.");        log.info("my first annotation job run, param: {},port:{}",param,port);        return ReturnT.SUCCESS;    }        public void init(){        log.info("my annotation job  init");    }    public void destroy(){        log.info("my job annotation job destory");    }
3.2 基于api任务
@Slf4jpublic class ApiJob extends IJobHandler {    @Override    public ReturnT execute(String param) throws Exception {        XxlJobLogger.log("XXL-JOB-API, Hello World.");        log.info("my job api run, param: {}",param);        return ReturnT.SUCCESS;    }}
 @PostConstruct    public void registerHandler(){   XxlJobExecutor.registJobHandler("myJobApiHandler",new ApiJob());    }
3.3 分片广播任务
 /**     * 2、分片任务     */    @XxlJob("myShardJobAnnotationHandler")    public ReturnT myShardJobAnnotationHandler(String param) throws Exception {        XxlJobLogger.log("XXL-JOB-ANNOTATION, myShardJobAnnotationHandler.");        log.info("my shard job run, param: {}",param);        ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();        log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());        // 业务逻辑        for (int i = 0; i < shardingVO.getTotal(); i++) {            if (i == shardingVO.getIndex()) {                log.info("第 {} 片, 命中分片开始处理", i);            } else {                log.info("第 {} 片, 忽略", i);            }        }        return ReturnT.SUCCESS;    }
上面是整理的比较实用的任务创建方式,个人偏好于注解形式,方法上加一个注解就完事了。
4.任务执行
剩下的就是傻白甜的界面操作了,走起。
4.1 单任务执行
创建一个路由策略为轮询的任务,指定corn表达式,并填入JobHandler为myJobAnnotationHandler,myJobAnnotationHandler其实就是spring IOC容器中管理bean的名称,有兴趣的童鞋可以看下源码。
为了演示效果,点击执行一次并进行任务参数输入。
轮询调用执行器服务效果如下:
4.2 子任务执行
更新任务,并指定子任务id为5,多个子任务的需要以逗号隔开
执行任务结果如下
4.3 分片广播任务执行
分片任务其实就是广播功能,每次触发,每个执行服务的业务执行类都会被调用,类似于kafka里面的不同消费组都要对同一个topic进行消费一样。
执行后的效果如下
太强势了,需要定时刷新项目中的配置信息,用这个方式很完美。
5.任务日志
任务日志其实是很重要的一块,方便回溯任务历史执行情况, 以便跟踪问题并矫正丢失的业务数据
查看调度备注,父子任务调度信息非常详细,子任务可以通过执行备注查看执行情况
查看控制台输出,里面的日志是执行器中XxlJobLogger类打印出来的
代码获取
1.访问地址：****https://github.com/pengziliu/GitHub-code-practice/
2.点击左下方阅读原文
总结
xxl-job还有很多特性,个人觉得实用的有任务超时控制/任务失败重试/任务失败告警 /邮件报警,这些都是与业务紧密相关的功能,能有效的规避生产事故而达到止损目的。
对比下之前使用过的任务调度组件,xxl-job将任务调度和执行进行解耦,大大提高了可用性和扩展性,代码的侵入性几乎没有,与spring boot无配置化,开箱即用的理念非常契合,任务调度平台可视化操作的确是爽爆了,感觉小白都可以用。

SpringBoot实战电商项目mall（40k+star）地址：…
摘要

之前我们讲过一个分布式任务调度框架PowerJob，可以通过可视化的方式来进行任务调度。但是有时候我们只是需要一个轻量级的任务调度功能，而PowerJob需要搭建调度中心未免有些重，这时候SpringBoot官方支持的任务调度框架Quartz就派上用场了！本文主要介绍Quartz在SpringBoot中的使用，让你在实现任务调度上有更多的选择！
Quartz简介

Quartz是一款功能强大的开源任务调度框架，几乎可以集成到任何Java应用程序中（小到单机应用，大到分布式应用）。Quartz可用于创建简单或复杂的任务调度，用以执行数以万计的任务。任务被定义为标准化的Java组件，Java编写的任务都可以被执行。
核心概念
  Quartz中有一些比较核心的概念，理解它们对使用Quartz很有帮助！
Scheduler（调度器）：Quartz中的任务调度器，通过Trigger和JobDetail可以用来调度、暂停和删除任务。
Trigger（触发器）：Quartz中的触发器，可以通过CRON表达式来指定任务执行的时间，时间到了会自动触发任务执行。
JobDetail（任务详情）：Quartz中需要执行的任务详情，包括了任务的唯一标识和具体要执行的任务，可以通过JobDataMap往任务中传递数据。
Job（任务）：Quartz中具体的任务，包含了执行任务的具体方法。
CRON表达式
Cron表达式是一个字符串，包括6~7个时间元素，在Quartz中可以用于指定任务的执行时间。
CRON的语法格式
Seconds Minutes Hours DayofMonth Month DayofWeek
CRON格式中每个时间元素的说明
时间元素
可出现的字符
有效数值范围
Seconds
, - * /
0-59
Minutes
, - * /
0-59
Hours
, - * /
0-23
DayofMonth
, - * / ? L W
0-31
Month
, - * /
1-12
DayofWeek
, - * / ? L #
1-7或SUN-SAT
CRON格式中特殊字符说明
 在线CRON表达式生成器
其实CRON表达式无需多记，需要使用的时候直接使用在线生成器就可以了，地址：
整合SpringBoot使用
接下来我们讲下如何在SpringBoot中使用Quartz来实现任务调度，在电商系统中往往会有需要定时发送邮件或者站内信的需求，我们以此为场景来实现下！
Quartz存储任务信息有两种方式，使用内存或者使用数据库来存储，这里我们采用数据库存储的方式，首先需要新建Quartz的相关表，建表脚本在项目的resources目录下，名称为，创建成功后数据库中多出11张表；
接下来在中添加Quartz的相关依赖即可，SpringBoot官方已经给我们提供好了相关Starter；
<!--SpringBoot集成QuartZ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
在中添加Quartz相关配置，配置说明直接看注释就好了，主要是对scheduler、jobStore和threadPool进行配置；
spring:
  quartz:
    job-store-type: jdbc # quartz任务存储类型：jdbc或memory
    wait-for-jobs-to-complete-on-shutdown: true # 关闭时等待任务完成
    overwrite-existing-jobs: true # 可以覆盖已有的任务
    properties: # quartz原生配置
      org:
        quartz:
          scheduler:
            instanceName: scheduler # 调度器实例名称
            instanceId: AUTO # 调度器实例ID自动生成
          jobStore:
            class: org.quartz.impl.jdbcjobstore.JobStoreTX # 调度信息存储处理类
            driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate # 使用完全兼容JDBC的驱动
            tablePrefix: QRTZ_ # quartz相关表前缀
            useProperties: false # 是否将JobDataMap中的属性转为字符串存储
          threadPool:
            class: org.quartz.simpl.SimpleThreadPool # 指定线程池实现类，对调度器提供固定大小的线程池
            threadCount: 10 # 设置并发线程数量
            threadPriority: 5 # 指定线程优先级
创建任务调度业务接口，定义好三个方法，分别为通过CRON表达式调度任务、调度指定时间的任务和取消定时任务；
/**
 * Quartz定时任务操作类
 * Created by macro on 2020/9/27.
 */
public interface ScheduleService {
    /**
     * 通过CRON表达式调度任务
     */
    String scheduleJob(Class<? extends Job> jobBeanClass, String cron, String data);

    /**
     * 调度指定时间的任务
     */
    String scheduleFixTimeJob(Class<? extends Job> jobBeanClass, Date startTime, String data);

    /**
     * 取消定时任务
     */
    Boolean cancelScheduleJob(String jobName);
}
创建任务调度业务实现类，通过Scheduler、CronTrigger、JobDetail的API实现相关方法；
/**
 * Quartz定时任务操作实现类
 * Created by macro on 2020/9/27.
 */
@Slf4j
@Service
public class ScheduleServiceImpl implements ScheduleService {
    @Autowired
    private Scheduler scheduler;
    private String defaultGroup = "default_group";

    @Override
    public String scheduleJob(Class<? extends Job> jobBeanClass, String cron, String data) {
        // 创建需要执行的任务
        String jobName = ().toString();
        JobDetail jobDetail = (jobBeanClass)
                .withIdentity(jobName, defaultGroup)
                .usingJobData("data", data)
                .build();
        //创建触发器，指定任务执行时间
        CronTrigger cronTrigger = ()
                .withIdentity(jobName, defaultGroup)
                .withSchedule((cron))
                .build();
        //使用调度器进行任务调度
        try {
            (jobDetail, cronTrigger);
        } catch (SchedulerException e) {
            ();
            ("创建定时任务失败！");
        }
        return jobName;
    }

    @Override
    public String scheduleFixTimeJob(Class<? extends Job> jobBeanClass, Date startTime, String data) {
        //日期转CRON表达式
        String startCron = ("%d %d %d %d %d ? %d",
                (startTime),
                (startTime),
                (startTime, true),
                (startTime),
                (startTime) + 1,
                (startTime));
        return scheduleJob(jobBeanClass, startCron, data);
    }

    @Override
    public Boolean cancelScheduleJob(String jobName) {
        boolean success = false;
        try {
            // 暂停触发器
            (new TriggerKey(jobName, defaultGroup));
            // 移除触发器中的任务
            (new TriggerKey(jobName, defaultGroup));
            // 删除任务
            (new JobKey(jobName, defaultGroup));
            success = true;
        } catch (SchedulerException e) {
            ();
        }
        return success;
    }
}
定义好需要执行的任务，继承QuartzJobBean类，实现executeInternal方法即可，这里定义了三个任务，定时发送邮件、定时发送站内信和执行CRON表达式任务；
/**
 * 发送邮件定时任务执行器
 * Created by macro on 2020/9/27.
 */
@Slf4j
@Component
public class SendEmailJob extends QuartzJobBean {
    @Autowired
    private ScheduleService scheduleService;

    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        Trigger trigger = ();
        JobDetail jobDetail = ();
        JobDataMap jobDataMap = ();
        String data = ("data");
        ("定时发送邮件操作：{}",data);
        //完成后删除触发器和任务
        (().getName());
    }
}
/**
 * 发送站内信定时任务执行器
 * Created by macro on 2020/9/27.
 */
@Slf4j
@Component
public class SendMessageJob extends QuartzJobBean {
    @Autowired
    private ScheduleService scheduleService;

    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        Trigger trigger = ();
        JobDetail jobDetail = ();
        JobDataMap jobDataMap = ();
        String data = ("data");
        ("定时发送站内信操作：{}",data);
        //完成后删除触发器和任务
        (().getName());
    }
}
/**
 * 使用CRON表达式的任务执行器
 * Created by macro on 2020/9/29.
 */
@Slf4j
@Component
public class CronProcessJob extends QuartzJobBean {
    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        JobDetail jobDetail = ();
        JobDataMap jobDataMap = ();
        String data = ("data");
        ("CRON表达式任务执行：{}",data);
    }
}
最后创建好任务调度相关接口，调用任务调度业务类即可。
/**
 * 定时任务调度相关接口
 * Created by macro on 2020/9/29.
 */
@Api(tags = "ScheduleController", description = "定时任务调度相关接口")
@RestController
@RequestMapping("/schedule")
public class ScheduleController {
    @Autowired
    private ScheduleService scheduleService;

    @ApiOperation("定时发送邮件")
    @PostMapping("/sendEmail")
    public CommonResult sendEmail(@RequestParam String startTime,@RequestParam String data) {
        Date date = (startTime, );
        String jobName = (SendEmailJob.class, date, data);
        return CommonResult.success(jobName);
    }

    @ApiOperation("定时发送站内信")
    @PostMapping("/sendMessage")
    public CommonResult sendMessage(@RequestParam String startTime,@RequestParam String data) {
        Date date = (startTime, );
        String jobName = (SendMessageJob.class, date, data);
        return CommonResult.success(jobName);
    }

    @ApiOperation("通过CRON表达式调度任务")
    @PostMapping("/scheduleJob")
    public CommonResult scheduleJob(@RequestParam String cron, @RequestParam String data) {
        String jobName = (CronProcessJob.class, cron, data);
        return CommonResult.success(jobName);
    }

    @ApiOperation("取消定时任务")
    @PostMapping("/cancelScheduleJob")
    public CommonResult cancelScheduleJob(@RequestParam String jobName) {
        Boolean success = (jobName);
        return CommonResult.success(success);
    }
}
运行测试
2020-09-30 11:23:00.035  INFO 10160 --- [eduler_Worker-1]      : 定时发送邮件操作：发送邮件内容
使用CRON表达式来启动一个定时任务，从0s开始，每隔10s执行一次；
2020-09-30 11:26:30.024  INFO 10160 --- [eduler_Worker-2] com.macro.mall.tiny.job.CronProcessJob   : CRON表达式任务执行：CRON消息内容
2020-09-30 11:26:40.025  INFO 10160 --- [eduler_Worker-3] com.macro.mall.tiny.job.CronProcessJob   : CRON表达式任务执行：CRON消息内容
2020-09-30 11:26:50.017  INFO 10160 --- [eduler_Worker-4] com.macro.mall.tiny.job.CronProcessJob   : CRON表达式任务执行：CRON消息内容
2020-09-30 11:27:00.023  INFO 10160 --- [eduler_Worker-5] com.macro.mall.tiny.job.CronProcessJob   : CRON表达式任务执行：CRON消息内容
2020-09-30 11:27:10.019  INFO 10160 --- [eduler_Worker-6] com.macro.mall.tiny.job.CronProcessJob   : CRON表达式任务执行：CRON消息内容
我们可以通过启动任务返回的jobName，调用取消定时任务的接口来取消任务，调用成功后定时任务不在执行。
参考资料
官方文档：…
作者：MacroZheng
链接：https://juejin.im/post/6885869364180942862
来源：掘金

前言
任务调度是java项目中常用的一种组件,可以指定任务在何时进行触发,最熟悉的是spring框架里面的quartz,较流行的有一些分布式调度组件,比如elastic-job/azkaban,都是基于quartz二次开发的,今天介绍一款分布式的任务调度器xxl-job。
项目介绍
xxl-job是一款极容易学习上手的轻量级开源分布式调度框架,分为管理端和执行端两块,管理端负责配置任务信息以及查看任务执行日志,执行端只需要配置与管理端的连接信息就可以进行具体的任务逻辑开发了,目前版本还在持续迭代中,使用简单,功能强大,具体功能特性可以看下官方介绍。废话不多说,直接进入实战把。
实战
1.服务端部署
从https://github.com/xuxueli/xxl-job下载项目,用mysql客户端工具Navicat执行项目根目录下doc/db/table_xxl_job.sql文件,库名自己可以自行修改,一共8张表,如下:

创建一个新的spring boot项目,将下载的xxl-job-admin目录下的文件以及pom.xml文件都拷贝到新建的项目中(如果不想新建项目可以直接用下载下来的项目进行修改部署),修改application.properties中的数据库连接信息。


小编是自己创建的新项目,需要手动改了pom.xml依赖xxl-job-core的版本为2.2.0


修改logback.xml中的日志输出路径。

好了,以上3步曲就搞定整个服务端配置了,启动项目,并访问http://localhost:8080/xxl-job-admin/ ,默认管理员账号admin/123456进行登录。

这交互,可以啊,是不是很带感。
2.执行端配置
创建一个新的module,跟服务端一样,也需要修改下logback.xml以及在pom.xml添加xxl-job-core的依赖。
为了模拟分布式效果,小编创建了2个配置文件来区分2个执行服务。
application-9998.properties
# web port
server.port=8081
# no web
#spring.main.web-environment=false
# log config
logging.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 token
xxl.job.accessToken=

### xxl-job executor appname
xxl.job.executor.appname=my-job-executor
### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is null
xxl.job.executor.address=
### xxl-job executor server-info
xxl.job.executor.ip=
xxl.job.executor.port=9998
### xxl-job executor log-path
xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler
### xxl-job executor log-retention-days
xxl.job.executor.logretentiondays=30


application-9999.properties
# web port
server.port=8082
# no web
#spring.main.web-environment=false

# log config
logging.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 token
xxl.job.accessToken=

### xxl-job executor appname
xxl.job.executor.appname=my-job-executor
### xxl-job executor registry-address: default use address to registry , otherwise use ip:port if address is null
xxl.job.executor.address=
### xxl-job executor server-info
xxl.job.executor.ip=
xxl.job.executor.port=9999
### xxl-job executor log-path
xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler
### xxl-job executor log-retention-days
xxl.job.executor.logretentiondays=30

细心的童鞋会发现只有server.port和xxl.job.executor.port不同,执行器服务跟spring boot一样,自带内嵌tomcat,也会暴露一个端口注册到服务端,进行高可用负载。
创建一个java config类,定义一个使用配置的XxlJobSpringExecutor执行类,如下
@Configuration
public 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);
        return xxlJobSpringExecutor;
    }

配置2个启动配置,分别启动,效果如下:


完美启动2个服务,看下服务端平台是不是有这两台执行服务的注册信息。

注意:为了演示,事先创建了一个执行器,AppName一定要与配置文件中xxl.job.executor.appname一致。
3.任务开发
3.1 基于方法注解任务
话不多说,直接上代码把,毕竟代码是程序员最好的交流方式。
    /**
     * 1、注解任务
     */
    @XxlJob(value = "myJobAnnotationHandler",init = "init", destroy = "destroy")
    public ReturnT<String> myJobAnnotationHandler(String param) throws Exception {
        XxlJobLogger.log("XXL-JOB-ANNOTATION, myJobAnnotationHandler.");

        log.info("my first annotation job run, param: {},port:{}",param,port);

        return ReturnT.SUCCESS;
    }
    
    public void init(){
        log.info("my annotation job  init");
    }

    public void destroy(){
        log.info("my job annotation job destory");
    }

3.2 基于api任务
@Slf4j
public class ApiJob extends IJobHandler {

    @Override
    public ReturnT<String> execute(String param) throws Exception {

        XxlJobLogger.log("XXL-JOB-API, Hello World.");

        log.info("my job api run, param: {}",param);

        return ReturnT.SUCCESS;
    }
}

 @PostConstruct
    public void registerHandler(){
   XxlJobExecutor.registJobHandler("myJobApiHandler",new ApiJob());
    }

3.3 分片广播任务
 /**
     * 2、分片任务
     */
    @XxlJob("myShardJobAnnotationHandler")
    public ReturnT<String> myShardJobAnnotationHandler(String param) throws Exception {
        XxlJobLogger.log("XXL-JOB-ANNOTATION, myShardJobAnnotationHandler.");

        log.info("my shard job run, param: {}",param);
        ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
        log.info("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());

        // 业务逻辑
        for (int i = 0; i < shardingVO.getTotal(); i++) {
            if (i == shardingVO.getIndex()) {
                log.info("第 {} 片, 命中分片开始处理", i);
            } else {
                log.info("第 {} 片, 忽略", i);
            }
        }
        return ReturnT.SUCCESS;
    }

上面是整理的比较实用的任务创建方式,个人偏好于注解形式,方法上加一个注解就完事了。
4.任务执行
剩下的就是傻白甜的界面操作了,走起。
4.1 单任务执行
创建一个路由策略为轮询的任务,指定corn表达式,并填入JobHandler为myJobAnnotationHandler,myJobAnnotationHandler其实就是spring IOC容器中管理bean的名称,有兴趣的童鞋可以看下源码。
为了演示效果,点击执行一次并进行任务参数输入。


轮询调用执行器服务效果如下:


4.2 子任务执行
更新任务,并指定子任务id为5,多个子任务的需要以逗号隔开

执行任务结果如下

4.3 分片广播任务执行
分片任务其实就是广播功能,每次触发,每个执行服务的业务执行类都会被调用,类似于kafka里面的不同消费组都要对同一个topic进行消费一样。

执行后的效果如下


太强势了,需要定时刷新项目中的配置信息,用这个方式很完美。
5.任务日志
任务日志其实是很重要的一块,方便回溯任务历史执行情况, 以便跟踪问题并矫正丢失的业务数据

查看调度备注,父子任务调度信息非常详细,子任务可以通过执行备注查看执行情况


查看控制台输出,里面的日志是执行器中XxlJobLogger类打印出来的

代码获取
**1.访问地址：**https://github.com/pengziliu/GitHub-code-practice/
2.点击左下方阅读原文
总结
xxl-job还有很多特性,个人觉得实用的有任务超时控制/任务失败重试/任务失败告警 /邮件报警,这些都是与业务紧密相关的功能,能有效的规避生产事故而达到止损目的。
对比下之前使用过的任务调度组件,xxl-job将任务调度和执行进行解耦,大大提高了可用性和扩展性,代码的侵入性几乎没有,与spring boot无配置化,开箱即用的理念非常契合,任务调度平台可视化操作的确是爽爆了,感觉小白都可以用。