1.异常处理与重启机制
1.对于chunk类型的Step,spring batch为我们提供了用于管理它的状态
2.状态的管理是通过ItemStream接口来实现的
3.ItemStream接口：
（1）open():每一次step执行会调用
（2）Update():每一个chunk去执行都会调用
（3）Close():所有的chunk执行完毕会调用
流程
构造例子
准备个cvs文件，在第33条数据，添加一条错误名字信息 ；当读取到这条数据时，抛出异常终止程序。
ItemReader测试代码
@Component("restartDemoReader")
public class RestartDemoReader implements ItemStreamReader<Customer> {
// 记录当前读取的行数
private Long curLine = 0L;
// 重启状态初始值
private boolean restart = false;
private FlatFileItemReader<Customer> reader = new FlatFileItemReader<>();
// 持久化信息到数据库
private ExecutionContext executionContext;
RestartDemoReader
public () {
reader.setResource(new ClassPathResource("restartDemo.csv"));
DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
tokenizer.setNames(new String[]{"id", "firstName", "lastName", "birthdate"});
DefaultLineMapper<Customer> lineMapper = new DefaultLineMapper<>();
lineMapper.setLineTokenizer(tokenizer);
lineMapper.setFieldSetMapper((fieldSet -> {
return Customer.builder().id(fieldSet.readLong("id"))
.firstName(fieldSet.readString("firstName"))
.lastName(fieldSet.readString("lastName"))
.birthdate(fieldSet.readString("birthdate"))
.build();
}));
lineMapper.afterPropertiesSet();
reader.setLineMapper(lineMapper);
}
@Override
public Customer read() throws Exception, UnexpectedInputException, ParseException,
NonTransientResourceException {
Customer customer = null;
this.curLine++;
//如果是重启，则从上一步读取的行数继续往下执行
if (restart) {
reader.setLinesToSkip(this.curLine.intValue()-1);
restart = false;
System.out.println("Start reading from line: " + this.curLine);
}
reader.open(this.executionContext);
customer = reader.read();
//当匹配到wrongName时，显示抛出异常，终止程序
if (customer != null) {
if (customer.getFirstName().equals("wrongName"))
throw new RuntimeException("Something wrong. Customer id: " + customer.getId());
} else {
curLine--;
}
return customer;
}
/**
* 判断是否是重启job
* @param executionContext
* @throws ItemStreamException
*/
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
this.executionContext = executionContext;
// 如果是重启job，从数据库读取重启的行数，从重启行数开始重新执行
if (executionContext.containsKey("curLine")) {
this.curLine = executionContext.getLong("curLine");
this.restart = true;
}
// 如果不是重启job，初始化行数，从第一行开始执行
else {
this.curLine = 0L;
executionContext.put("curLine", this.curLine.intValue());
}
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
// 每执行完一个批次chunk，打印当前行数
System.out.println("update curLine: " + this.curLine);
executionContext.put("curLine", this.curLine);
}
@Override
public void close() throws ItemStreamException {
}
}
Job配置
以10条记录为一个批次，进行读取
@Configuration
public class RestartDemoJobConfiguration {
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Autowired
@Qualifier("flatFileDemoWriter")
private ItemWriter<? super Customer> flatFileDemoWriter;
@Autowired
@Qualifier("restartDemoReader")
private ItemReader<Customer> restartDemoReader;
@Bean
public Job restartDemoJob(){
return jobBuilderFactory.get("restartDemoJob")
.start(restartDemoStep())
.build();
}
@Bean
public Step restartDemoStep() {
return stepBuilderFactory.get("restartDemoStep")
.<Customer,Customer>chunk(10)
.reader(restartDemoReader)
.writer(flatFileDemoWriter)
.build();
}
}
程序执行时会经历一下三个步骤演示异常处理机制
1.当第一次执行时，程序在33行抛出异常异常，curline值是30；
这时，可以查询数据库 batch_step_excution表，发现curline值已经以 键值对形式，持久化进数据库（上文以10条数据为一个批次；故33条数据异常时，curline值为30）
2.接下来，更新wrongName，再次执行程序；
3.程序会执行open方法，判断数据库step中map是否存在curline，如果存在，则是重跑，即读取curline，从该批次开始往下继续执行；
2.容错机制
Spring batch的容错机制是一种与事务机制相结合的机制，它主要包括有3种操作：
- restart
- restart是针对job来使用，是重启job的一个操作。默认情况下，当任务出现异常时，SpringBatch会结束任务，当使用相同参数重启任务时，SpringBatch会去执行未执行的剩余任务
- retry
- retry是对job的某一step而言，处理一条数据item的时候发现有异常，则重试一次该数据item的step的操作。
- skip
- skip是对job的某一个step而言，处理一条数据item的时候发现有异常，则跳过该数据item的step的操作。
restart示例代码如下，当第一次执行的时候，上下文中没有该字段，抛出异常，第二次执行，已存在该字段，执行成功
retry、skip示例如下，更改一下之前step的配置，参考代码如下：
@Bean
public Step stepForTranscation(StepBuilderFactory stepBuilderFactory, @Qualifier("stepForTranscationReader")ListItemReader<String> reader,
@Qualifier("stepForTranscationProcessor")ItemProcessor<String, String> processor, @Qualifier("stepForTranscationWriter")ItemWriter<String> writer) {
return stepBuilderFactory.get("stepForTranscation")
.<String, String> chunk(3)
.reader(reader)
.processor(processor)
.writer(writer).
faultTolerant().
retryLimit(3).
retry(DataIntegrityViolationException.class).
skipLimit(1).
skip(DataIntegrityViolationException.class).
startLimit(3)
.build();
}
这里设置了允许重试的次数为3次，允许跳过的数据最多为1条，如果job失败了，运行重跑次数最多为3次。
在skip后面配置跳过错误的监听器SkipListener
public class Mr implements SkipListener<String, String>{ // 发生读操作跳过错误时，需要执行的监听 public void onSkipInRead(Throwable t){ }
// 发生写操作跳过错误时，需要执行的监听
public void onSkipInWrite(String item, Throwable t){
}
// 处理数据时跳过错误时，需要执行的监听
public void onSkipInProcess (String item, Throwable t){
System.out.println(item + "occur exception" + t);
}
}
重新运行程序，可以得到新的结果：
结果
这次可以看到，12条数据中总共有11条数据进入到数据库，而过长的008008008008数据，则因为设置了skip，所以容错机制允许它不进入数据库，这次的Spring batch最终没有因为回滚而中断。
查阅一下Spring batch的持久化数据表：
可以看出，的确是有一条数据被跳过了，但因为是我们允许它跳过的，所以整个job顺利完成，即COMPLETED。
