spring batch 之 step总结

吕亮

2023-12-01

在spring batch中 job是step运行的框架，而step则是运行具体业务的步骤，以下是step相关的一些总结。

step结构

一个Step通常涵盖三个部分：读数据（Reader）、处理数据（Processor）和写数据（Writer）。但是并不是所有的Step都需要自身来完成数据的处理，比如存储过程等方式是通过外部功能来完成，因此Spring Batch提供了2种Step的处理方式：
1）面向分片的ChunkStep，
2）面向过程的TaskletStep。
一般使用ChunkStep。

ChunkStep

在Step中数据是按记录（按行）处理的，但是每条记录处理完毕之后马上提交事物反而会导致IO的巨大压力。因此Spring Batch提供了数据处理的分片功能。设置了分片之后，一次工作会从Read开始，然后交由给Processor处理。处理完毕后会进行聚合，待聚合到一定的数量的数据之后一次性调用Write将数据提交到物理数据库。
如果在聚合数据期间出现任何错误，所有的这些数据都将不执行写入。

代码示例

@Bean
public Step testStep(PlatformTransactionManager transactionManager) {
	return stepBuilderFactory.get("testStep")
				.transactionManager(transactionManager)
				.<String, String>chunk(int) 
				.reader(testReader()) 
				.processor(testProcessor)
				.writer(testWriter()) 
				.build();
}

transactionManager：使用默认的 PlatformTransactionManager 对事物进行管理。当配置好事物之后Spring Batch会自动对事物进行管理，无需开发人员显示操作。
processor看具体需求可以不写

提交间隔

Step使用PlatformTransactionManager管理事物。每次事物提交的间隔根据chunk方法中配置的数据执行。提交间隔设置太小，那么会浪费需要多不必要的资源，提交间隔设置的太长，会导致事物链太长占用空间，并且出现失败会导致大量数据回滚。一般设为10到20。

step重启次数

某些Step可能用于处理一些先决的任务，所以当Job再次重启时这Step就没必要再执行，可以通过设置startLimit来限定某个Step重启的次数。当设置为1时候表示仅仅运行一次，而出现重启时将不再执行。

stepBuilderFactory.get("testStep")
	.<String, String>chunk(int)
	.reader(testReader()) 
	.processor(testProcessor)
	.writer(testWriter()) 
	.startLimit(1)
	.build();

step每次重启都运行

可以通过设置allow-start-if-complete为true告知框架每次重启该Step都要执行

stepBuilderFactory.get("testStep")
	 .<String, String>chunk(int)
	 .reader(testReader()) 
	 .processor(testProcessor)
	 .writer(testWriter()) 
	 .allowStartIfComplete(true)
	 .build();

step失败后跳过

stepBuilderFactory.get("testStep")
 	 .<String, String>chunk(int)
 	 .reader(testReader()) 
	 .processor(testProcessor)
 	 .writer(testWriter()) 
 	 .skipLimit(10)
	 .skip(Exception.class)
	 .noSkip(FileNotFoundException.class)
 	 .build();

skip-limit（skipLimit方法）配置的参数表示当跳过的次数超过数值时则会导致整个Step失败，从而停止继续运行
skip表示要当捕捉到Exception异常就跳过。但是Exception有很多继承类，此时可以使用noSkip方法指定某些异常不能跳过。

step重试

stepBuilderFactory.get("testStep")
 	 .<String, String>chunk(int)
  	 .reader(testReader()) 
  	 .processor(testProcessor)
   	 .writer(testWriter()) 
	 .faultTolerant()
	 .retryLimit(3)
	 .retry(DeadlockLoserDataAccessException.class)
   	 .build();

retry(DeadlockLoserDataAccessException.class)表示只有捕捉到该异常才会重试，retryLimit(3)表示最多重试3次，faultTolerant()表示启用对应的容错功能。

step控制不会滚

stepBuilderFactory.get("testStep")
	 .<String, String>chunk(int)
   	 .reader(testReader()) 
   	 .processor(testProcessor)
         .writer(testWriter()) 
  	 .faultTolerant()
	 .noRollback(ValidationException.class) //不必回滚的异常
    	 .build();

noRollback属性为Step提供了不必进行事物回滚的异常配置

step数据重读

stepBuilderFactory.get("testStep")
 	 .<String, String>chunk(int)
     	 .reader(testReader()) 
    	 .processor(testProcessor)
         .writer(testWriter()) 
	 .readerIsTransactionalQueue() //数据重读
         .build();

默认情况下如果错误不是发生在Reader阶段，那么没必要再去重新读取一次数据。但是某些场景下需要Reader部分也需要重新执行，比如Reader是从一个JMS队列中消费消息，当发生回滚的时候消息也会在队列上重放，因此也要将Reader纳入到回滚的事物中，根据这个场景可以使用readerIsTransactionalQueue来配置数据重读

step事务属性

	//配置事物属性
	DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
	attribute.setPropagationBehavior(Propagation.REQUIRED.value());
	attribute.setIsolationLevel(Isolation.DEFAULT.value());
	attribute.setTimeout(30);

	return this.stepBuilderFactory.get("testStep")
				.<String, String>chunk(int)
      				 .reader(testReader()) 
     				 .processor(testProcessor)
        			 .writer(testWriter()) 
				.transactionAttribute(attribute) //设置事物属性
				.build();

Step注册 ItemStream

ItemStream是用于每一个阶段（Reader、Processor、Writer）的“生命周期回调数据处理器”。在4.×版本之后默认注入注册了通用的ItemStream。
一是使用stream方法

stepBuilderFactory.get("testStep")
	 .<String, String>chunk(int)
	 .reader(testReader()) 
         .processor(testProcessor)
         .writer(testWriter()) 
	 .stream(fileItemWriter1())
	 .build();

二是使用相关方法的代理

	List<ItemWriter> writers = new ArrayList<>(2);
	writers.add(fileItemWriter1());
	writers.add(fileItemWriter2());
	CompositeItemWriter itemWriter = new CompositeItemWriter();
	itemWriter.setDelegates(writers);
	return itemWriter;

StepExecution拦截器

在Step执行的过程中会产生各种各样的事件，开发人员可以利用各种Listener接口对Step及Item进行监听。通常在创建一个Step的时候添加拦截器

stepBuilderFactory.get("testStep")
	 .<String, String>chunk(int)
  	 .reader(testReader()) 
         .processor(testProcessor)
         .writer(testWriter()) 
 	 .listener(chunkListener()) //添加拦截器
	 .build();

StepExecutionListener

StepExecutionListener可以看做一个通用的Step拦截器，他的作用是在Step开始之前和结束之后进行拦截处理

public interface StepExecutionListener extends StepListener {
    void beforeStep(StepExecution stepExecution); //Step执行之前
    ExitStatus afterStep(StepExecution stepExecution); //Step执行完毕之后
}

ChunkListener

ChunkListener是在数据事物发生的两端被触发。chunk的配置决定了处理多少项记录才进行一次事物提交，ChunkListener的作用就是对一次事物开始之后或事物提交之后进行拦截

public interface ChunkListener extends StepListener {
    void beforeChunk(ChunkContext context); //事物开始之后，ItemReader调用之前
    void afterChunk(ChunkContext context); //事物提交之后
    void afterChunkError(ChunkContext context); //事物回滚之后
}

ItemReadListener

该接口用于对Reader相关的事件进行监控

public interface ItemReadListener<T> extends StepListener {
    void beforeRead();
    void afterRead(T item);
    void onReadError(Exception ex);
}

beforeRead在每次Reader调用之前被调用，afterRead在每次Reader成功返回之后被调用，而onReadError会在出现异常之后被调用，可以将其用于记录异常日志

ItemProcessListener

ItemProcessListener和ItemReadListener类似，是围绕着ItemProcessor进行处理的

public interface ItemProcessListener<T, S> extends StepListener {
    void beforeProcess(T item); //processor执行之前
    void afterProcess(T item, S result); //processor直线成功之后
    void onProcessError(T item, Exception e); //processor执行出现异常
}

ItemWriteListener

ItemWriteListener的功能和ItemReadListener、ItemReadListener类似，但是需要注意的是它接收和处理的数据对象是一个List。List的长度与chunk配置相关

public interface ItemWriteListener<S> extends StepListener {
    void beforeWrite(List<? extends S> items);
    void afterWrite(List<? extends S> items);
    void onWriteError(Exception exception, List<? extends S> items);
}

SkipListener

ItemReadListener、ItemProcessListener和ItemWriteListener都提供了错误拦截处理的机制，但是没有处理跳过（skip）的数据记录。因此框架提供了SkipListener来专门处理那么被跳过的记录

public interface SkipListener<T,S> extends StepListener {
    void onSkipInRead(Throwable t); //Read期间导致跳过的异常
    void onSkipInProcess(T item, Throwable t); //Process期间导致跳过的异常
    void onSkipInWrite(S item, Throwable t); //Write期间导致跳过的异常
}

SkipListener的价值是可以将那些未能成功处理的记录在某个位置保存下来，然后交给其他批处理进一步解决，或者人工来处理。Spring Batch保证以下2个特征：

跳过的元素只会出现一次。
SkipListener始终在事物提交之前被调用，这样可以保证监听器使用的事物资源不会被业务事物影响。

TaskletStep

TaskletStep是一个非常简单的接口，仅有一个方法——execute。TaskletStep会反复的调用这个方法直到获取一个RepeatStatus.FINISHED返回或者抛出一个异常。所有的Tasklet调用都会包装在一个事物中。

stepBuilderFactory.get("step1")
			.tasklet(myTasklet()) //注入Tasklet的实现
    			.build();

step顺序执行

jobBuilderFactory.get("job")
		.start(stepA())
		.next(stepB()) //顺序执行
		.next(stepC())
		.build();

step条件执行

jobBuilderFactory.get("job")
				.start(stepA()) //启动时执行的step
				.on("*").to(stepB()) //默认跳转到stepB
				.from(stepA()).on("FAILED").to(stepC()) //当返回的ExitStatus为"FAILED"时，执行。
				.end()
				.build();

Step的停机退出机制

Spring Batch为Job提供了三种退出机制，这些机制为批处理的执行提供了丰富的控制方法。在介绍退出机制之前需要回顾一下数据批处理概念一文中关于StepExecution的内容。在StepExecution中有2个表示状态的值，一个名为status，另外一个名为exitStatus。前者也被称为BatchStatus。前面以及介绍了ExitStatus的使用，他可以控制Step执行链条的条件执行过程。除此之外BatchStatus也会参与到过程的控制。

End退出

默认情况下（没有使用end、fail方法结束），Job要顺序执行直到退出，这个退出称为end。这个时候，BatchStatus=COMPLETED、ExitStatus=COMPLETED，表示成功执行。除了Step链式处理自然退出，也可以显示调用end来退出系统。

jobBuilderFactory.get("job")
		.start(step1()) //启动
		.next(step2()) //顺序执行
		.on("FAILED").end()
		.from(step2()).on("*").to(step3()) //条件执行
		.end()
		.build();

step1到step2是顺序执行，当step2的exitStatus返回"FAILED"时则直接End退出。其他情况执行Step3。

Fail退出

除了end还可以使用fail退出，这个时候，BatchStatus=FAILED、ExitStatus=EARLY TERMINATION，表示执行失败。这个状态与End最大的区别是Job会尝试重启执行新的JobExecution。

jobBuilderFactory.get("job")
		.start(step1()) //执行step1
		.next(step2()).on("FAILED").fail() //step2的ExitStatus=FAILED 执行fail
		.from(step2()).on("*").to(step3()) //否则执行step3
		.end()
		.build();

在指定的节点中断

Spring Batch还支持在指定的节点退出，退出后下次重启会从中断的点继续执行。中断的作用是某些批处理到某个步骤后需要人工干预，当干预完之后又接着处理

jobBuilderFactory.get("job")
 		//如果step1的ExitStatus=COMPLETED则在step2中断
		.start(step1()).on("COMPLETED").stopAndRestart(step2())
		 //否则直接退出批处理
		.end()
		.build();

程序化流程的分支

可以直接进行编码来控制Step之间的扭转，Spring Batch提供了JobExecutionDecider接口来协助分支管理

public class MyDecider implements JobExecutionDecider {
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        String status;
        if (someCondition()) {
            status = "FAILED";
        }
        else {
            status = "COMPLETED";
        }
        return new FlowExecutionStatus(status);
    }
}

接着将MyDecider作为过滤器添加到配置过程中

jobBuilderFactory.get("job")
		.start(step1())
		.next(decider()).on("FAILED").to(step2())
		.from(decider()).on("COMPLETED").to(step3())
		.end()
		.build();

流程分裂

在线性处理过程中，流程都是一个接着一个执行的。但是为了满足某些特殊的需要，Spring Batch提供了执行的过程分裂并行Step的方法。

@Bean
public Job job() {
	Flow flow1 = new FlowBuilder<SimpleFlow>("flow1")
			.start(step1())
			.next(step2())
			.build();//并行流程1
	Flow flow2 = new FlowBuilder<SimpleFlow>("flow2")
			.start(step3())
			.build();//并行流程2

	return this.jobBuilderFactory.get("job")
				.start(flow1)
				.split(new SimpleAsyncTaskExecutor()) //创建一个异步执行任务
				.add(flow2)
				.next(step4()) //2个分支执行完毕之后再执行step4。
				.end()
				.build();
}

这里表示flow1和flow2会并行执行，待2者执行成功后执行step4。

数据绑定

在Job或Step的任何位置，都可以获取到统一配置的数据。比如使用标准的Spring Framework方式

@Bean
public FlatFileItemReader flatFileItemReader(@Value("${input.file.name}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

当我们通过配置文件（application.properties中 input.file.name=filepath）或者jvm参数（-Dinput.file.name=filepath）指定某些数据时，都可以通过这种方式获取到对应的配置参数。此外，也可以从JobParameters从获取到Job运行的上下文参数

@StepScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters['input.file.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

无论是JobExecution还是StepExecution，其中的内容都可以通过这种方式去获取参数

@StepScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{jobExecutionContext['input.file.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

@StepScope
@Bean
public FlatFileItemReader flatFileItemReader(@Value("#{stepExecutionContext['input.file.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

Job Scope

Job Scope的概念和 Step Scope类似，都是用于标识在到了某个执行时间段再添加和注入Bean。@JobScope用于告知框架知道JobInstance存在时候才初始化对应的@Bean

@JobScope
@Bean
// 初始化获取 jobParameters中的参数
public FlatFileItemReader flatFileItemReader(@Value("#{jobParameters[input]}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}

@JobScope
@Bean
// 初始化获取jobExecutionContext中的参数
public FlatFileItemReader flatFileItemReader(@Value("#{jobExecutionContext中的参数['input.name']}") String name) {
	return new FlatFileItemReaderBuilder<Foo>()
			.name("flatFileItemReader")
			.resource(new FileSystemResource(name))
			...
}