调度与监控-spring batch(7)结合xxl-job进行批处理

tags: springbatch

1.引言

经过前面几篇文章对Spring Batch的介绍，同时结合示例，从最简单的helloworld字符串输出，到读取文件到数据库的数据同步，然后是数据库到数据库，接着结合BeetlSql进一步简化数据库读写，再通过动态参数绑定实现增量同步，由浅到深，已经可以基本满足数据抽取，数据同步的工作了，下面是之前的文章列表：

批处理任务，一般做为后台服务定时运行，服务运维工作则是批处理服务之外的一项重要工作，包括服务运行如何监控，任务运行统计，任务执行日志查看等等。前面已说过，Spring Batch是批处理框架，并不是调度框架。调度框架,简单点可以使用quartz，crontab，而市面上已经有比较多功能完善的调度框架，经过比较，个人感觉xxl-job是比较容易上手，而且功能比较齐全，因此，这里介绍一下，使用xxl-job对Spring Batch的批处理任务进行调度，实现监控功能。

2.xxl-job介绍

xxl-job是一个轻量级的分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。系统架构设计合理，把调度系统与执行器分离，调度系统负责调度相关的逻辑，执行器则是具体任务的实现逻辑，开发者可以直接使用调度系统，然后实现自己任务逻辑来做为执行器即可，做到开箱即用。具体使用方式及详细说明，可见其官方文档

按照xxl-job的文档，先安装xxl-job的数据库，然后把xxl-job-admin启动起来，启动起来的页面如下：

xxl-job界面

3.编写Spring Batch执行器

本示例是基于上一篇文章所用到的数据库增量同步的示例，只需要在原来的基础上添加一个执行器即可，本示例为spring-batch-xxl-executor。按照xxl-job的官方文档，里面有说明如何配置部署“执行器项目”。具体到本实例，配置，实现如下：

3.1 添加maven依赖

添加xxl-job的核心依赖，如下：

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

3.2 添加执行器配置文件`executor.properties`

执行器配置需要填写包括执行器名称、调度系统地址等，如下：

### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
xxl.job.admin.addresses=http://127.0.0.1:8089/xxl-job-admin

### xxl-job executor address
xxl.job.executor.appname=${spring.application.name}
xxl.job.executor.ip=
xxl.job.executor.port=9999

### xxl-job, access token
xxl.job.accessToken=

### xxl-job log path
xxl.job.executor.logpath=logs
### xxl-job log retention days
xxl.job.executor.logretentiondays=-1

3.3 执行器组件配置

添加java配置文件，获取以上的executor.properties配置。如下：

@Configuration
@ConfigurationProperties(prefix = "xxl.job")
@PropertySource("classpath:/config/executor.properties")
@Slf4j
public class JobExecutorConfig {
    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;
    
    ......略

    @Bean(initMethod = "start", destroyMethod = "destroy")
    public XxlJobSpringExecutor xxlJobExecutor() {
        log.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppName(appName);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }
}

3.4 编写执行器

执行器，其实就是继承了IJobHandler类的实现，在execute方法实现任务实现逻辑即可。本示例中，只需要按之前测试的启动任务的逻辑实现即可。如下：

@JobHandler(value="incrementUserJobHandler")
@Component
public class JobIncrementUserHandler extends IJobHandler {
    @Autowired
    private JobLauncherService jobLauncherService;

    @Autowired
    private IncrementService incrementService;

    @Autowired
    private Job incrementJob;
    @Override
    public ReturnT<String> execute(String s) throws Exception {
        return JobUtil.runJob4Executor("incrementUser",incrementService,jobLauncherService,incrementJob);
    }
}

4.使用xxl-job进行Spring Batch任务调度

把xxl-job-admin和spring-batch-xxl-executor进行部署后（注意配置正确），就可以在xxl-job-admin界面中对任务进行调度。具体添加流程可参照xxl-job文档即可。简单说就是添加执行器spring-batch-xxl-executor，然后添加任务incrementUserJobHandler，指定执行的cron表达式，启动即可。执行任务后输出如下：

执行结果

5. 总结

本文基于数据库增量同步的示例，使用Spring Batch结合xxl-job，实现可监控的任务调度，使得批处理任务可维护性更强，希望需要使用Spring Batch进行批处理任务开发的人员有帮助。

java 16

book 4

SpringBatch 9

docsify 2

github 2

MongoDB 1

troubleshooting 8

monitor 8

jvm 1

jvisualvm 1

jconsole 1

mat 1

btrace 1

arthas 1

ai 1

machinelearning 1

deeplearning 1

springboot 8

swagger 2

doc 3

logback 2

log 2

apache 1

https 1

dynamic-proxy 1

aop 1

life 3

summary 3

multi-datasource 3

minio 2

static-resource-server 1

distributed 1

storage 1

python 3

os 1

path 1

excel 2

openpyxl 1

db 1

mysql 1