2024最新!一文看懂Spring Batch批处理(大白话版,干货满满)
一、课程目标
课程目标
-
系统了解Spring Batch批处理
-
项目中能熟练使用Spring Batch批处理
课程内容
前置知识
- Java基础
- Maven
- Spring SpringMVC SpringBoot
- MyBatis
适合人群
- 想学习的所有人
二、Spring Batch简介
2.1 何为批处理？
何为批处理，大白话：就是将数据分批次进行处理的过程。比如：银行对账逻辑，跨系统数据同步等。
常规的批处理操作步骤：系统A从数据库中导出数据到文件，系统B读取文件数据并写入到数据库
典型批处理特点：
-
自动执行，根据系统设定的工作步骤自动完成
-
数据量大，少则百万，多则上千万甚至上亿。(如果是10亿，100亿那只能上大数据了)
-
定时执行，比如：每天，每周，每月执行。
2.2 Spring Batch了解
官网介绍：https://docs.spring.io/spring-batch/docs/current/reference/html/spring-batch-intro.html#spring-batch-intro
这里挑重点讲下：
- Sping Batch 是一个轻量级的、完善的的批处理框架，旨在帮助企业建立健壮、高效的批处理应用。
- Spring Batch 是Spring的一个子项目，基于Spring框架为基础的开发的框架
- Spring Batch 提供大量可重用的组件，比如：日志，追踪，事务，任务作业统计，任务重启，跳过，重复，资源管理等
- Spring Batch 是一个批处理应用框架，不提供调度框架，如果需要定时处理需要额外引入-调度框架，比如： Quartz
2.3 Spring Batch 优势
Spring Batch 框架通过提供丰富的开箱即用的组件和高可靠性、高扩展性的能力，使得开发批处理应用的人员专注于业务处理，提高处理应用的开发能力。下面就是使用Spring Batch后能获取到优势：
-
丰富的开箱即用组件
-
面向Chunk的处理
-
事务管理能力
-
元数据管理
-
易监控的批处理应用
-
丰富的流程定义
-
健壮的批处理应用
-
易扩展的批处理应用
-
复用企业现有的IT代码
2.4 Spring Batch 架构
Spring Batch 核心架构分三层：应用层，核心层，基础架构层。
Application：应用层，包含所有的批处理作业，程序员自定义代码实现逻辑。
Batch Core：核心层，包含Spring Batch启动和控制所需要的核心类，比如：JobLauncher， Job，Step等。
Batch Infrastructure：基础架构层，提供通用的读，写与服务处理。
三层体系使得Spring Batch 架构可以在不同层面进行扩展，避免影响，实现高内聚低耦合设计。
三、入门案例
3.1 批量处理流程
前面对Spring Batch 有大体了解之后，那么开始写个案例玩一下。
开始前，先了解一下Spring Batch程序运行大纲：
JobLauncher：作业调度器，作业启动主要入口。
Job：作业，需要执行的任务逻辑，
Step：作业步骤，一个Job作业由1个或者多个Step组成，完成所有Step操作，一个完整Job才算执行结束。
ItemReader：Step步骤执行过程中数据输入。可以从数据源(文件系统，数据库，队列等)中读取Item(数据记录)。
ItemWriter：Step步骤执行过程中数据输出，将Item(数据记录)写入数据源(文件系统，数据库，队列等)。
ItemProcessor：Item数据加工逻辑(输入)，比如：数据清洗，数据转换，数据过滤，数据校验等
JobRepository： 保存Job或者检索Job的信息。SpringBatch需要持久化Job(可以选择数据库/内存)，JobRepository就是持久化的接口
3.2 入门案例-H2版(内存)
需求：打印一个hello spring batch！不带读/写/处理
步骤1：导入依赖
org.springframework.boot spring-boot-starter-parent 2.7.3 org.springframework.boot spring-boot-starter-batch org.projectlombok lombok com.h2database h2 runtime org.springframework.boot spring-boot-starter-test其中的h2是一个嵌入式内存数据库，后续可以使用MySQL替换
步骤2：创建测试方法
package com.langfeiyes.batch._01_hello; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class HelloJob { //job调度器 @Autowired private JobLauncher jobLauncher; //job构造器工厂 @Autowired private JobBuilderFactory jobBuilderFactory; //step构造器工厂 @Autowired private StepBuilderFactory stepBuilderFactory; //任务-step执行逻辑由tasklet完成 @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Hello SpringBatch...."); return RepeatStatus.FINISHED; } }; } //作业步骤-不带读/写/处理 @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //定义作业 @Bean public Job job(){ return jobBuilderFactory.get("hello-job") .start(step1()) .build(); } public static void main(String[] args) { SpringApplication.run(HelloJob.class, args); } }步骤3：分析
例子是一个简单的SpringBatch 入门案例，使用了最简单的一种步骤处理模型：Tasklet模型，step1中没有带上读/写/处理逻辑，只有简单打印操作，后续随学习深入，我们再讲解更复杂化模型。
3.3 入门案例-MySQL版
MySQL跟上面的h2一样，区别在连接数据库不一致。
步骤1：在H2版本基础上导入MySQL依赖
mysql mysql-connector-java 8.0.12步骤2：配置数据库四要素与初始化SQL脚本
spring: datasource: username: root password: admin url: jdbc:mysql://127.0.0.1:3306/springbatch?serverTimezone=GMT%2B8&useSSL=false&allowPublicKeyRetrieval=true driver-class-name: com.mysql.cj.jdbc.Driver # 初始化数据库，文件在依赖jar包中 sql: init: schema-locations: classpath:org/springframework/batch/core/schema-mysql.sql mode: always #mode: never这里要注意， sql.init.model 第一次启动为always， 后面启动需要改为never，否则每次执行SQL都会异常。
第一次启动会自动执行指定的脚本，后续不需要再初始化
步骤3：测试
跟H2版一样。
四、入门案例解析
1>@EnableBatchProcessing
批处理启动注解，要求贴配置类或者启动类上
@SpringBootApplication @EnableBatchProcessing public class HelloJob { ... }贴上@EnableBatchProcessing注解后，SpringBoot会自动加载JobLauncher JobBuilderFactory StepBuilderFactory 类并创建对象交给容器管理，要使用时，直接@Autowired即可
//job调度器 @Autowired private JobLauncher jobLauncher; //job构造器工厂 @Autowired private JobBuilderFactory jobBuilderFactory; //step构造器工厂 @Autowired private StepBuilderFactory stepBuilderFactory;
2>配置数据库四要素
批处理允许重复执行，异常重试，此时需要保存批处理状态与数据，Spring Batch 将数据缓存在H2内存中或者缓存在指定数据库中。入门案例如果要保存在MySQL中，所以需要配置数据库四要素。
3>创建Tasklet对象
//任务-step执行逻辑由tasklet完成 @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Hello SpringBatch...."); return RepeatStatus.FINISHED; } }; }Tasklet负责批处理step步骤中具体业务执行，它是一个接口，有且只有一个execute方法，用于定制step执行逻辑。
public interface Tasklet { RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception; }execute方法返回值是一个状态枚举类：RepeatStatus，里面有可继续执行态与已经完成态
public enum RepeatStatus { /** * 可继续执行的-tasklet返回这个状态会进入***循环 */ CONTINUABLE(true), /** * 已经完成态 */ FINISHED(false); .... }4>创建Step对象
//作业步骤-不带读/写/处理 @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); }Job作业执行靠Step步骤执行，入门案例选用最简单的Tasklet模式，后续再讲Chunk块处理模式。
5>创建Job并执行Job
//定义作业 @Bean public Job job(){ return jobBuilderFactory.get("hello-job") .start(step1()) .build(); }创建Job对象交给容器管理，当springboot启动之后，会自动去从容器中加载Job对象，并将Job对象交给JobLauncherApplicationRunner类，再借助JobLauncher类实现job执行。
验证过程；
打断点，debug模式启动
SpringApplication类run方法
JobLauncherApplicationRunner类
JobLauncher接口–实现类：SimpleJobLauncher
五、作业对象 Job
5.1 作业介绍
5.1.1 作业定义
Job作业可以简单理解为一段业务流程的实现，可以根据业务逻辑拆分一个或者多个逻辑块(step)，然后业务逻辑顺序，逐一执行。
所以作业可以定义为：能从头到尾独立执行的有序的步骤(Step)列表。
-
有序的步骤列表
一次作业由不同的步骤组成，这些步骤顺序是有意义的，如果不按照顺序执行，会引起逻辑混乱，比如购物结算，先点结算，再支付，最后物流，如果反过来那就乱套了，作业也是这么一回事。
-
从头到尾
一次作业步骤固定了，在没有外部交互情况下，会从头到尾执行，前一个步骤做完才会到后一个步骤执行，不允许随意跳转，但是可以按照一定逻辑跳转。
-
独立
每一个批处理作业都应该不受外部依赖影响情况下执行。
看回这幅图，批处理作业Job是由一组步骤Step对象组成，每一个作业都有自己名称，可以定义Step执行顺序。
5.1.2 作业代码设计
前面定义讲了作业执行是相互独立的，代码该怎么设计才能保证每次作业独立的性呢？
答案是：Job instance(作业实例) 与 Job Execution(作业执行对象)
Job instance(作业实例)
当作业运行时，会创建一个Job Instance(作业实例)，它代表作业的一次逻辑运行，可通过作业名称与作业标识参数进行区分。
比如一个业务需求： 每天定期数据同步，作业名称-daily-sync-job 作业标记参数-当天时间
Job Execution(作业执行对象)
当作业运行时，也会创建一个Job Execution(作业执行器)，负责记录Job执行情况(比如：开始执行时间，结束时间，处理状态等)。
那为啥会出现上面架构设计呢？原因：批处理执行过程中可能出现两种情况：
-
一种是一次成功
仅一次就成从头到尾正常执行完毕，在数据库中会记录一条Job Instance 信息， 跟一条 Job Execution 信息
-
另外一种异常执行
在执行过程因异常导致作业结束，在数据库中会记录一条Job Instance 信息， 跟一条Job Execution 信息。如果此时使用相同识别参数再次启动作业，那么数据库中不会多一条Job Instance 信息， 但是会多了一条Job Execution 信息，这就意味中任务重复执行了。刚刚说每天批处理任务案例，如果当天执行出异常，那么人工干预修复之后，可以再次执行。
最后来个总结：
Job Instance = Job名称 + 识别参数
Job Instance 一次执行创建一个 Job Execution对象
完整的一次Job Instance 执行可能创建一个Job Execution对象，也可能创建多个Job Execution对象
5.2 作业配置
再看回入门案例
package com.langfeiyes.batch._01_hello; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class HelloJob { //job构造器工厂 @Autowired private JobBuilderFactory jobBuilderFactory; //step构造器工厂 @Autowired private StepBuilderFactory stepBuilderFactory; //任务-step执行逻辑由tasklet完成 @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Hello SpringBatch...."); return RepeatStatus.FINISHED; } }; } //作业步骤-不带读/写/处理 @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //定义作业 @Bean public Job job(){ return jobBuilderFactory.get("hello-job") .start(step1()) .build(); } public static void main(String[] args) { SpringApplication.run(HelloJob.class, args); } }在启动类中贴上@EnableBatchProcessing注解，SpringBoot会自动听JobLauncher JobBuilderFactory StepBuilderFactory 对象，分别用于执行Jog，创建Job，创建Step逻辑。有了这些逻辑，Job批处理就剩下组装了。
5.3 作业参数
5.3.1 JobParameters
前面提到，作业的启动条件是作业名称 + 识别参数，Spring Batch使用JobParameters类来封装了所有传给作业参数。
我们看下JobParameters 源码
public class JobParameters implements Serializable { private final Map parameters; public JobParameters() { this.parameters = new LinkedHashMap(); } public JobParameters(Map parameters) { this.parameters = new LinkedHashMap(parameters); } ..... }从上面代码/截图来看，JobParameters 类底层维护了Map，是一个Map集合的封装器，提供了不同类型的get操作。
5.3.2 作业参数设置
还记得Spring Batch 入门案例吗，当初debug时候看到Job作业最终是调用时 **JobLauncher **(job启动器)接口run方法启动。
看下源码：JobLauncher
public interface JobLauncher { public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException; }在JobLauncher 启动器执行run方法时，直接传入即可。
jobLauncher.run(job, params);
那我们使用SpringBoot 方式启动Spring Batch该怎么传值呢？
1>定义ParamJob类，准备好要执行的job
package com.langfeiyes.batch._02_params; import org.springframework.batch.core.*; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class ParamJob { //job构造器工厂 @Autowired private JobBuilderFactory jobBuilderFactory; //step构造器工厂 @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("param SpringBatch...."); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("param-job") .start(step1()) .build(); } public static void main(String[] args) { SpringApplication.run(HelloJob.class, args); } }2>使用idea的命令传值的方式设置job作业参数
注意：如果不想这么麻烦，其实也可以，先空参数执行一次，然后指定参数后再执行。
点击绿色按钮，启动SpringBoot程序，作业运行之后，会在batch_job_execution_params 增加一条记录，用于区分唯一的Job Instance实例
注意：如果不改动JobParameters 参数内容，再执行一次批处理，会直接报错。
原因：Spring Batch 相同Job名与相同标识参数只能成功执行一次。
5.3.3 作业参数获取
当将作业参数传入到作业流程，该如何获取呢？
Spring Batch 提供了2种方案：
方案1：使用ChunkContext类
ParamJob类中tasklet写法
@Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---name:" + parameters.get("name")); return RepeatStatus.FINISHED; } }; }注意：job名：param-job job参数：name=dafei 已经执行了，再执行会报错
所以要么改名字，要么改参数，这里选择改job名字（拷贝一份job实例方法，然后注释掉，修改Job名称）
// @Bean // public Job job(){ // return jobBuilderFactory.get("param-job") // .start(step1()) // .build(); // } @Bean public Job job(){ return jobBuilderFactory.get("param-chunk-job") .start(step1()) .build(); }方案2：使用@Value 延时获取
@StepScope @Bean public Tasklet tasklet(@Value("#{jobParameters['name']}")String name){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("params---name:" + name); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet(null)) .build(); }step1调用tasklet实例方法时不需要传任何参数，Spring Boot 在加载Tasklet Bean实例时会自动注入。
// @Bean // public Job job(){ // return jobBuilderFactory.get("param-chunk-job") // .start(step1()) // .build(); // } @Bean public Job job(){ return jobBuilderFactory.get("param-value-job") .start(step1()) .build(); }这里要注意，必须贴上@StepScope ，表示在启动项目的时候，不加载该Step步骤bean，等step1()被调用时才加载。这就是所谓延时获取。
5.3.4 作业参数校验
当外部传入的参数进入作业时，如何确保参数符合期望呢？使用Spring Batch 的参数校验器：JobParametersValidator 接口。
先来看下JobParametersValidator 接口源码：
public interface JobParametersValidator { void validate(@Nullable JobParameters parameters) throws JobParametersInvalidException; }JobParametersValidator 接口有且仅有唯一的validate方法，参数为JobParameters，没有返回值。这就意味着不符合参数要求，需要抛出异常来结束步骤。
定制参数校验器
Spring Batch 提供JobParametersValidator参数校验接口，其目的就是让我们通过实现接口方式定制参数校验逻辑。
需求：如果传入作业的参数name值 为null 或者 “” 时报错
public class NameParamValidator implements JobParametersValidator { @Override public void validate(JobParameters parameters) throws JobParametersInvalidException { String name = parameters.getString("name"); if(!StringUtils.hasText(name)){ throw new JobParametersInvalidException("name 参数不能为空"); } } }其中的JobParametersInvalidException 异常是Spring Batch 专门提供参数校验失败异常，当然我们也可以自定义或使用其他异常。
package com.langfeiyes.batch._03_param_validator; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Map; @SpringBootApplication @EnableBatchProcessing public class ParamValidatorJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---name:" + parameters.get("name")); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //配置name参数校验器 @Bean public NameParamValidator validator(){ return new NameParamValidator(); } @Bean public Job job(){ return jobBuilderFactory.get("name-param-validator-job") .start(step1()) .validator(validator()) //参数校验器 .build(); } public static void main(String[] args) { SpringApplication.run(ParamValidatorJob.class, args); } }新定义validator()实例方法，将定制的参数解析器加到Spring容器中，修改job()实例方法，加上.validator(validator()) 校验逻辑。
第一次启动时，没有传任何参数
String name = parameters.getString("name");name为null，直接报错
加上name=dafei参数之后，正常执行
默认参数校验器
除去上面的定制参数校验器外，Spring Batch 也提供2个默认参数校验器：DefaultJobParametersValidator(默认参数校验器) 跟 CompositeJobParametersValidator(组合参数校验器)。
DefaultJobParametersValidator参数校验器
public class DefaultJobParametersValidator implements JobParametersValidator, InitializingBean { private Collection requiredKeys; private Collection optionalKeys; .... }默认的参数校验器它功能相对简单，维护2个key集合requiredKeys 跟 optionalKeys
- requiredKeys 是一个集合，表示作业参数jobParameters中必须包含集合中指定的keys
- optionalKeys 也是一个集合，该集合中的key 是可选参数
需求：如果作业参数没有name参数报错，age参数可有可无
package com.langfeiyes.batch._03_param_validator; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.job.DefaultJobParametersValidator; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Map; @SpringBootApplication @EnableBatchProcessing public class ParamValidatorJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---name:" + parameters.get("name")); System.out.println("params---age:" + parameters.get("age")); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //配置name参数校验器 @Bean public NameParamValidator validator(){ return new NameParamValidator(); } //配置默认参数校验器 @Bean public DefaultJobParametersValidator defaultValidator(){ DefaultJobParametersValidator defaultValidator = new DefaultJobParametersValidator(); defaultValidator.setRequiredKeys(new String[]{"name"}); //必填 defaultValidator.setOptionalKeys(new String[]{"age"}); //可选 return defaultValidator; } @Bean public Job job(){ return jobBuilderFactory.get("default-param-validator-job") .start(step1()) //.validator(validator()) //参数校验器 .validator(defaultValidator()) //默认参数校验器 .build(); } public static void main(String[] args) { SpringApplication.run(ParamValidatorJob.class, args); } }新定义defaultValidator() 实例方法，将默认参数解析器加到Spring容器中，修改job实例方法，加上**.validator(defaultValidator())。**
右键启动，不填name 跟 age 参数，直接报错
如果填上name参数，即使不填age参数，可以通过，原因是age是可选的。
组合参数校验器
CompositeJobParametersValidator 组合参数校验器，顾名思义就是将多个参数校验器组合在一起。
看源码，大体能看出该校验器逻辑
public class CompositeJobParametersValidator implements JobParametersValidator, InitializingBean { private List validators; @Override public void validate(@Nullable JobParameters parameters) throws JobParametersInvalidException { for (JobParametersValidator validator : validators) { validator.validate(parameters); } } public void setValidators(List validators) { this.validators = validators; } @Override public void afterPropertiesSet() throws Exception { Assert.notNull(validators, "The 'validators' may not be null"); Assert.notEmpty(validators, "The 'validators' may not be empty"); } }底层维护一个validators 集合，校验时调用validate 方法，依次执行校验器集合中校验器方法。另外，多了一个afterPropertiesSet方法，用于校验validators 集合中的校验器是否为null。
需求：要求步骤中必须有name属性，并且不能为空
分析：必须有，使用DefaultJobParametersValidator 参数校验器， 不能为null，使用指定定义的NameParamValidator参数校验器
package com.langfeiyes.batch._03_param_validator; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.job.CompositeJobParametersValidator; import org.springframework.batch.core.job.DefaultJobParametersValidator; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Arrays; import java.util.Map; @SpringBootApplication @EnableBatchProcessing public class ParamValidatorJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---name:" + parameters.get("name")); System.out.println("params---age:" + parameters.get("age")); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //配置name参数校验器 @Bean public NameParamValidator validator(){ return new NameParamValidator(); } //配置默认参数校验器 @Bean public DefaultJobParametersValidator defaultValidator(){ DefaultJobParametersValidator defaultValidator = new DefaultJobParametersValidator(); defaultValidator.setRequiredKeys(new String[]{"name"}); //必填 defaultValidator.setOptionalKeys(new String[]{"age"}); //可选 return defaultValidator; } //配置组合参数校验器 @Bean public CompositeJobParametersValidator compositeValidator(){ DefaultJobParametersValidator defaultValidator = new DefaultJobParametersValidator(); defaultValidator.setRequiredKeys(new String[]{"name"}); //name必填 defaultValidator.setOptionalKeys(new String[]{"age"}); //age可选 NameParamValidator nameParamValidator = new NameParamValidator(); //name 不能为空 CompositeJobParametersValidator compositeValidator = new CompositeJobParametersValidator(); //按照传入的顺序，先执行defaultValidator 后执行nameParamValidator compositeValidator.setValidators(Arrays.asList(defaultValidator, nameParamValidator)); try { compositeValidator.afterPropertiesSet(); //判断校验器是否为null } catch (Exception e) { e.printStackTrace(); } return compositeValidator; } @Bean public Job job(){ return jobBuilderFactory.get("composite-param-validator-job") .start(step1()) //.validator(validator()) //参数校验器 //.validator(defaultValidator()) //默认参数校验器 .validator(compositeValidator()) //组合参数校验器 .build(); } public static void main(String[] args) { SpringApplication.run(ParamValidatorJob.class, args); } }新定义compositeValidator() 实例方法，将组合参数解析器加到spring容器中，修改job()实例方法，加上**.validator(compositeValidator())。**
右键启动，不填name参数，测试报错。如果放开name参数，传null值，一样报错。
5.3.5 作业增量参数
不知道大家发现了没有，每次运行作业时，都改动作业名字，或者改动作业的参数，原因是作业启动有限制：相同标识参数与相同作业名的作业，只能成功运行一次。那如果想每次启动，又不想改动标识参数跟作业名怎么办呢？答案是：使用JobParametersIncrementer (作业参数增量器)
看下源码，了解一下原理
public interface JobParametersIncrementer { JobParameters getNext(@Nullable JobParameters parameters); }JobParametersIncrementer 增量器是一个接口，里面只有getNext方法，参数是JobParameters 返回值也是JobParameters。通过这个getNext方法，在作业启动时我们可以给JobParameters 添加或者修改参数。简单理解就是让标识参数每次都变动
作业递增run.id参数
Spring Batch 提供一个run.id自增参数增量器：RunIdIncrementer，每次启动时，里面维护名为run.id 标识参数，每次启动让其自增 1。
看下源码：
public class RunIdIncrementer implements JobParametersIncrementer { private static String RUN_ID_KEY = "run.id"; private String key = RUN_ID_KEY; public void setKey(String key) { this.key = key; } @Override public JobParameters getNext(@Nullable JobParameters parameters) { JobParameters params = (parameters == null) ? new JobParameters() : parameters; JobParameter runIdParameter = params.getParameters().get(this.key); long id = 1; if (runIdParameter != null) { try { id = Long.parseLong(runIdParameter.getValue().toString()) + 1; } catch (NumberFormatException exception) { throw new IllegalArgumentException("Invalid value for parameter " + this.key, exception); } } return new JobParametersBuilder(params).addLong(this.key, id).toJobParameters(); } }核心getNext方法，在JobParameters 对象维护一个run.id，每次作业启动时，都调用getNext方法获取JobParameters，保证其 run.id 参数能自增1
具体用法：
package com.langfeiyes.batch._04_param_incr; import com.langfeiyes.batch._03_param_validator.NameParamValidator; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.job.CompositeJobParametersValidator; import org.springframework.batch.core.job.DefaultJobParametersValidator; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Arrays; import java.util.Map; @SpringBootApplication @EnableBatchProcessing public class IncrementParamJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---run.id:" + parameters.get("run.id")); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("incr-params-job") .start(step1()) .incrementer(new RunIdIncrementer()) //参数增量器(run.id自增) .build(); } public static void main(String[] args) { SpringApplication.run(IncrementParamJob.class, args); } }修改tasklet()方法，获取run.id参数，修改job实例方法，加上**.incrementer(new RunIdIncrementer())** ，保证参数能自增。
连续执行3次，观察：batch_job_execution_params 表
其中的run.id参数值一直增加，其中再多遍也没啥问题。
作业时间戳参数
run.id 作为标识参数貌似没有具体业务意义，如果将时间戳作为标识参数那就不一样了，比如这种运用场景：每日任务批处理，这时就需要记录每天的执行时间了。那该怎么实现呢？
Spring Batch 中没有现成时间戳增量器，需要自己定义
//时间戳作业参数增量器 public class DailyTimestampParamIncrementer implements JobParametersIncrementer { @Override public JobParameters getNext(JobParameters parameters) { return new JobParametersBuilder(parameters) .addLong("daily", new Date().getTime()) //添加时间戳 .toJobParameters(); } }定义一个标识参数：daily，记录当前时间戳
package com.langfeiyes.batch._04_param_incr; import com.langfeiyes.batch._03_param_validator.NameParamValidator; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.core.job.CompositeJobParametersValidator; import org.springframework.batch.core.job.DefaultJobParametersValidator; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Arrays; import java.util.Map; @SpringBootApplication @EnableBatchProcessing public class IncrementParamJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { Map parameters = chunkContext.getStepContext().getJobParameters(); System.out.println("params---daily:" + parameters.get("daily")); return RepeatStatus.FINISHED; } }; } //时间戳增量器 @Bean public DailyTimestampParamIncrementer dailyTimestampParamIncrementer(){ return new DailyTimestampParamIncrementer(); } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("incr-params-job") .start(step1()) //.incrementer(new RunIdIncrementer()) //参数增量器(run.id自增) .incrementer(dailyTimestampParamIncrementer()) //时间戳增量器 .build(); } public static void main(String[] args) { SpringApplication.run(IncrementParamJob.class, args); } }定义实例方法dailyTimestampParamIncrementer()将自定义时间戳增量器添加Spring容器中，修改job()实例方法，添加.incrementer(dailyTimestampParamIncrementer()) 增量器，修改tasklet() 方法，获取 daily参数。
连续执行3次，查看batch_job_execution_params 表
很明显可以看出daily在变化，而run.id 没有动，是3，为啥？因为**.incrementer(new RunIdIncrementer())** 被注释掉了。
5.4 作业监听器
作业监听器：用于监听作业的执行过程逻辑。在作业执行前，执行后2个时间点嵌入业务逻辑。
- 执行前：一般用于初始化操作， 作业执行前需要着手准备工作，比如：各种连接建立，线程池初始化等。
- 执行后：业务执行完后，需要做各种清理动作，比如释放资源等。
Spring Batch 使用JobExecutionListener 接口 实现作业监听。
public interface JobExecutionListener { //作业执行前 void beforeJob(JobExecution jobExecution); //作业执行后 void afterJob(JobExecution jobExecution); }需求：记录作业执行前，执行中，与执行后的状态
方式一：接口方式
//作业状态--接口方式 public class JobStateListener implements JobExecutionListener { //作业执行前 @Override public void beforeJob(JobExecution jobExecution) { System.err.println("执行前-status：" + jobExecution.getStatus()); } //作业执行后 @Override public void afterJob(JobExecution jobExecution) { System.err.println("执行后-status：" + jobExecution.getStatus()); } }定义JobStateListener 实现JobExecutionListener 接口，重写beforeJob，afterJob 2个方法。
import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class StatusListenerJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { JobExecution jobExecution = contribution.getStepExecution().getJobExecution(); System.err.println("执行中-status：" + jobExecution.getStatus()); return RepeatStatus.FINISHED; } }; } //状态监听器 @Bean public JobStateListener jobStateListener(){ return new JobStateListener(); } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("status-listener-job") .start(step1()) .listener(jobStateListener()) //设置状态监听器 .build(); } public static void main(String[] args) { SpringApplication.run(StatusListenerJob.class, args); } }新加jobStateListener()实例方法创建对象交个Spring容器管理，修改job()方法，添加.listener(jobStateListener()) 状态监听器，直接执行，观察结果
方式二：注解方式
除去上面通过实现接口方式实现监听之外，也可以使用**@BeforeJob @AfterJob** 2个注解实现
//作业状态--注解方式 public class JobStateAnnoListener { @BeforeJob public void beforeJob(JobExecution jobExecution) { System.err.println("执行前-anno-status：" + jobExecution.getStatus()); } @AfterJob public void afterJob(JobExecution jobExecution) { System.err.println("执行后-anno-status：" + jobExecution.getStatus()); } }import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.listener.JobListenerFactoryBean; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class StatusListenerJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { JobExecution jobExecution = contribution.getStepExecution().getJobExecution(); System.err.println("执行中-anno-status：" + jobExecution.getStatus()); return RepeatStatus.FINISHED; } }; } //状态监听器 /* @Bean public JobStateListener jobStateListener(){ return new JobStateListener(); }*/ @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("status-listener-job1") .start(step1()) .incrementer(new RunIdIncrementer()) //.listener(jobStateListener()) //设置状态监听器 .listener(JobListenerFactoryBean.getListener(new JobStateAnnoListener())) .build(); } public static void main(String[] args) { SpringApplication.run(StatusListenerJob.class, args); } }修改job()方法，添加**.listener(JobListenerFactoryBean.getListener(new JobStateAnnoListener()))**状态监听器，直接执行，观察结果
不需要纠结那一长串方法是啥逻辑，只需要知道它能将指定监听器对象加载到spring容器中。
5.5 执行上下文
5.5.1 作业与步骤上下文
语文中有个词叫上下文，比如：联系上下文解读一下作者所有表达意思。从这看上下文有环境，语境，氛围的意思。类比到编程，业内也喜欢使用Context表示上下文。比如Spring容器： SpringApplicationContext 。有上下文这个铺垫之后，我们来看下Spring Batch的上下文。
Spring Batch 有2个比较重要的上下文：
-
JobContext
JobContext 绑定 JobExecution 执行对象为Job作业执行提供执行环境(上下文)。
作用：维护JobExecution 对象，实现作业收尾工作，与处理各种作业回调逻辑
-
StepContext
StepContext 绑定 StepExecution 执行对象为Step步骤执行提供执行环境(上下文)。
作用：维护StepExecution 对象，实现步骤收尾工作，与处理各种步骤回调逻辑
5.5.2 执行上下文
除了上面讲的JobContext 作业上下文， StepContext 步骤上线下文外，还有Spring Batch还维护另外一个上下文：ExecutionContext 执行上下文，作用是：数据共享
Spring Batch 中 ExecutionContext 分2大类
-
Job ExecutionContext
作用域：一次作业运行，所有Step步骤间数据共享。
-
Step ExecutionContext：
作用域：一次步骤运行，单个Step步骤间(ItemReader/ItemProcessor/ItemWrite组件间)数据共享。
5.5.3 作业与步骤执行链
5.5.4 作业与步骤引用链
-
作业线
Job—JobInstance—JobContext—JobExecution–ExecutionContext
-
步骤线
Step–StepContext –StepExecution–ExecutionContext
5.5.5 作业上下文API
JobContext context = JobSynchronizationManager.getContext(); JobExecution jobExecution = context.getJobExecution(); Map jobParameters = context.getJobParameters(); Map jobExecutionContext = context.getJobExecutionContext();
5.5.6 步骤上下文API
ChunkContext chunkContext = xxx; StepContext stepContext = chunkContext.getStepContext(); StepExecution stepExecution = stepContext.getStepExecution(); Map stepExecutionContext = stepContext.getStepExecutionContext(); Map jobExecutionContext = stepContext.getJobExecutionContext();
5.5.7 执行上下文API
ChunkContext chunkContext = xxx; //步骤 StepContext stepContext = chunkContext.getStepContext(); StepExecution stepExecution = stepContext.getStepExecution(); ExecutionContext executionContext = stepExecution.getExecutionContext(); executionContext.put("key", "value"); //------------------------------------------------------------------------- //作业 JobExecution jobExecution = stepExecution.getJobExecution(); ExecutionContext executionContext = jobExecution.getExecutionContext(); executionContext.put("key", "value");5.5.8 API综合小案例
需求：观察作业ExecutionContext与 步骤ExecutionContext数据共享
分析：
1>定义step1 与step2 2个步骤
2>在step1中设置数据
作业-ExecutionContext 添加： key-step1-job value-step1-job
步骤-ExecutionContext 添加： key-step1-step value-step1-step
3>在step2中打印观察
作业-ExecutionContext 步骤-ExecutionContext
package com.langfeiyes.batch._06_context; import com.langfeiyes.batch._04_param_incr.DailyTimestampParamIncrementer; import org.springframework.batch.core.*; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.listener.JobListenerFactoryBean; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.scope.context.JobContext; import org.springframework.batch.core.scope.context.JobSynchronizationManager; import org.springframework.batch.core.scope.context.StepContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; @SpringBootApplication @EnableBatchProcessing public class ExecutionContextJob { @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet1(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { //步骤 ExecutionContext stepEC = chunkContext.getStepContext().getStepExecution().getExecutionContext(); stepEC.put("key-step1-step","value-step1-step"); System.out.println("------------------1---------------------------"); //作业 ExecutionContext jobEC = chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext(); jobEC.put("key-step1-job","value-step1-job"); return RepeatStatus.FINISHED; } }; } @Bean public Tasklet tasklet2(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { //步骤 ExecutionContext stepEC = chunkContext.getStepContext().getStepExecution().getExecutionContext(); System.err.println(stepEC.get("key-step1-step")); System.out.println("------------------2---------------------------"); //作业 ExecutionContext jobEC = chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext(); System.err.println(jobEC.get("key-step1-job")); return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet1()) .build(); } @Bean public Step step2(){ return stepBuilderFactory.get("step2") .tasklet(tasklet2()) .build(); } @Bean public Job job(){ return jobBuilderFactory.get("execution-context-job") .start(step1()) .next(step2()) .incrementer(new RunIdIncrementer()) .build(); } public static void main(String[] args) { SpringApplication.run(ExecutionContextJob.class, args); } }运行结果：
可以看出，在stepContext 设置的参数作用域仅在StepExecution 执行范围有效，而JobContext 设置参数作用与在所有StepExcution 有效，有点局部与全局 的意思。
打开数据库观察表：batch_job_execution_context 跟 batch_step_execution_context 表
JobContext数据保存到：batch_job_execution_context
StepContext数据保存到：batch_step_execution_context
总结：
步骤数据保存在Step ExecutionContext，只能在Step中使用，作业数据保存在Job ExecutionContext，可以在所有Step中共享
六、步骤对象 Step
前面一章节讲完了作业的相关介绍，本章节重点讲解步骤。
6.1 步骤介绍
一般认为步骤是一个独立功能组件，因为它包含了一个工作单元需要的所有内容，比如：输入模块，输出模块，数据处理模块等。这种设计好处在哪？给开发者带来更自由的操作空间。
目前Spring Batch 支持2种步骤处理模式：
-
简单具于Tasklet 处理模式
这种模式相对简单，前面讲的都是居于这个模式批处理
@Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("Hello SpringBatch...."); return RepeatStatus.FINISHED; } }; }只需要实现Tasklet接口，就可以构建一个step代码块。循环执行step逻辑，直到tasklet.execute方法返回RepeatStatus.FINISHED
-
居于块(chunk)的处理模式
居于块的步骤一般包含2个或者3个组件：1>ItemReader 2>ItemProcessor(可选) 3>ItemWriter 。 当用上这些组件之后，Spring Batch 会按块处理数据。
6.2 简单Tasklet
学到这，我们写过很多简单Tasklet模式步骤，但是都没有深入了解过，这节就细致分析一下具有Tasklet 步骤使用。
先看下Tasklet源码
public interface Tasklet { @Nullable RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception; }Tasklet 接口有且仅有一个方法：execute，
参数有2个：
StepContribution：步骤信息对象，用于保存当前步骤执行情况信息，核心用法：设置步骤结果状态contribution.setExitStatus(ExitStatus status)
contribution.setExitStatus(ExitStatus.COMPLETED);
ChunkContext：chuck上下文，跟之前学的StepContext JobContext一样，区别是它用于记录chunk块执行场景。通过它可以获取前面2个对象。
返回值1个：
RepeatStatus：当前步骤状态， 它是枚举类，有2个值，一个表示execute方法可以循环执行，一个表示已经执行结束。
public enum RepeatStatus { /** * 当前步骤依然可以执行，如果步骤返回该值，会一直循环执行 */ CONTINUABLE(true), /** * 当前步骤结束，可以为成功也可以表示不成，仅代表当前step执行结束了 */ FINISHED(false); }需求：练习上面RepeatStatus状态
@SpringBootApplication @EnableBatchProcessing public class SimpleTaskletJob { @Autowired private JobLauncher jobLauncher; @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public Tasklet tasklet(){ return new Tasklet() { @Override public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { System.out.println("------>" + System.currentTimeMillis()); //return RepeatStatus.CONTINUABLE; //循环执行 return RepeatStatus.FINISHED; } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .tasklet(tasklet()) .build(); } //定义作业 @Bean public Job job(){ return jobBuilderFactory.get("step-simple-tasklet-job") .start(step1()) .build(); } public static void main(String[] args) { SpringApplication.run(SimpleTaskletJob.class, args); } }6.3 居于块Tasklet
居于块的Tasklet相对简单Tasklet来说，多了3个模块：ItemReader( 读模块)， ItemProcessor(处理模块)，ItemWriter(写模块)， 跟它们名字一样， 一个负责数据读， 一个负责数据加工，一个负责数据写。
结构图：
时序图：
需求：简单演示chunk Tasklet使用
ItemReader ItemProcessor ItemWriter 都接口，直接使用匿名内部类方式方便创建
package com.langfeiyes.batch._08_step_chunk_tasklet; import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.*; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.Arrays; import java.util.List; @SpringBootApplication @EnableBatchProcessing public class ChunkTaskletJob { @Autowired private JobLauncher jobLauncher; @Autowired private JobBuilderFactory jobBuilderFactory; @Autowired private StepBuilderFactory stepBuilderFactory; @Bean public ItemReader itemReader(){ return new ItemReader() { @Override public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { System.out.println("-------------read------------"); return "read-ret"; } }; } @Bean public ItemProcessor itemProcessor(){ return new ItemProcessor() { @Override public Object process(Object item) throws Exception { System.out.println("-------------process------------>" + item); return "process-ret->" + item; } }; } @Bean public ItemWriter itemWriter(){ return new ItemWriter() { @Override public void write(List items) throws Exception { System.out.println(items); } }; } @Bean public Step step1(){ return stepBuilderFactory.get("step1") .chunk(3) //设置块的size为3次 .reader(itemReader()) .processor(itemProcessor()) .writer(itemWriter()) .build(); } //定义作业 @Bean public Job job(){ return jobBuilderFactory.get("step-chunk-tasklet-job") .start(step1()) .incrementer(new RunIdIncrementer()) .build(); } public static void main(String[] args) { SpringApplication.run(ChunkTaskletJob.class, args); } }执行完了之后结果
-------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret -------------process------------>read-ret -------------process------------>read-ret [process-ret->read-ret, process-ret->read-ret, process-ret->read-ret] -------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret -------------process------------>read-ret -------------process------------>read-ret [process-ret->read-ret, process-ret->read-ret, process-ret->read-ret] -------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret -------------process------------>read-ret -------------process------------>read-ret [process-ret->read-ret, process-ret->read-ret, process-ret->read-ret] ....
观察上面打印结果，得出2个得出。
1>程序一直在循环打印，先循环打印3次reader， 再循环打印3次processor，最后一次性输出3个值。
2>***循环重复上面步骤
问题来了，为啥会出现这种效果，该怎么改进？
其实这个是ChunkTasklet 执行特点，ItemReader会一直循环读，直到返回null，才停止。而processor也是一样，itemReader读多少次，它处理多少次， itemWriter 一次性输出当前次输入的所有数据。
我们改进一下上面案例，要求只读3次， 只需要改动itemReader方法就行
int timer = 3; @Bean public ItemReader itemReader(){ return new ItemReader() { @Override public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { if(timer > 0){ System.out.println("-------------read------------"); return "read-ret-" + timer--; }else{ return null; } } }; }结果不在***循环了
-------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret-3 -------------process------------>read-ret-2 -------------process------------>read-ret-1 [process-ret->read-ret-3, process-ret->read-ret-2, process-ret->read-ret-1]
思考一个问题， 如果将timer改为 10，而 .chunk(3) 不变结果会怎样？
-------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret-10 -------------process------------>read-ret-9 -------------process------------>read-ret-8 [process-ret->read-ret-10, process-ret->read-ret-9, process-ret->read-ret-8] -------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret-7 -------------process------------>read-ret-6 -------------process------------>read-ret-5 [process-ret->read-ret-7, process-ret->read-ret-6, process-ret->read-ret-5] -------------read------------ -------------read------------ -------------read------------ -------------process------------>read-ret-4 -------------process------------>read-ret-3 -------------process------------>read-ret-2 [process-ret->read-ret-4, process-ret->read-ret-3, process-ret->read-ret-2] -------------read------------ -------------process------------>read-ret-1 [process-ret->read-ret-1]
找出规律了嘛？
当chunkSize = 3 表示 reader 先读3次，提交给processor处理3次，最后由writer输出3个值
timer =10， 表示数据有10条，一个批次(趟)只能处理3条数据，需要4个批次(趟)来处理。
是不是有批处理味道出来
结论：chunkSize 表示： 一趟需要ItemReader读多少次，ItemProcessor要处理多少次。
ChunkTasklet 泛型
上面案例默认的是使用Object类型读、写、处理数据，如果明确了Item的数据类型，可以明确指定具体操作泛型。
import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.*; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import java.util.List; //开启 spring batch 注解--可以让spring容器创建springbatch操作相关类对象 @EnableBatchProcessing //springboot 项目，启动注解， 保证当前为为启动类 @SpringBootApplication public class ChunkTaskletJob { //作业启动器 @Autowired private JobLauncher jobLauncher; //job构造工厂---用于构建job对象 @Autowired private JobBuilderFactory jobBuilderFactory; //step 构造工厂--用于构造step对象 @Autowired private StepBuilderFactory stepBuilderFactory; int timer = 10; //读操作 @Bean public ItemReader itemReader(){ return new ItemReader() { @Override public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { if(timer > 0){ System.out.println("-------------read------------"); return "read-ret-->" + timer--; }else{ return null; } } }; } //处理操作 @Bean public ItemProcessor itemProcessor(){ return new ItemProcessor() { @Override public String process(String item) throws Exception { System.out.println("-------------process------------>" + item); return "process-ret->" + item; } }; } //写操作 @Bean public ItemWriter itemWriter(){ return new ItemWriter() { @Override public void write(List
-
-
-
-
-
-
-
-
- 想学习的所有人
免责声明：我们致力于保护作者版权，注重分享，被刊用文章因无法核实真实出处，未能及时与作者取得联系，或有版权异议的，请联系管理员，我们会立即处理!
部分文章是来自自研大数据AI进行生成,内容摘自(百度百科,百度知道,头条百科,中国民法典,刑法,牛津词典,新华词典,汉语词典,国家院校,科普平台)等数据,内容仅供学习参考,不准确地方联系删除处理!
图片声明：本站部分配图来自人工智能系统AI生成,觅知网授权图片,PxHere摄影无版权图库和百度，360，搜狗等多加搜索引擎自动关键词搜索配图，如有侵权的图片，请第一时间联系我们，邮箱：ciyunidc@ciyunshuju.com。本站只作为美观性配图使用,无任何非法侵犯第三方意图,一切解释权归图片著作权方,本站不承担任何责任。如有恶意碰瓷者,必当奉陪到底严惩不贷!
