基于Spring|基于Spring Batch 配置重试逻辑基于SpringBatch配置重试逻辑

1. 应用示例

批处理应用读取csv文件
处理类如下
最终输出结果为

2. 给处理增加重试功能

因此我们配置批处理job在失败的情况下重试三次

3. 测试重试功能

第三次成功调用
job成功执行

4. 总结

Spring Batch在处理过程中遇到错误job默认会执行失败。为了提高应用程序的健壮性，我们需要处理临时异常造成失败。本文我们探讨如何配置Spring Batch的重试逻辑。

1. 应用示例
批处理应用读取csv文件

sammy, 1234, 31/10/2015, 10000john, 9999, 3/12/2015, 12321

然后，通过调用rest接口处理每条记录，获取用户的年龄和邮编属性，为了正确输出日期，可以在属性上增加@XmlJavaTypeAdapter(LocalDateTimeAdapter.class)注解：

@XmlRootElement(name = “transactionRecord”)@Datapublic class Transaction {private String username; private int userId; private int age; private String postCode; private LocalDateTime transactionDate; private double amount; }

处理类如下

public class RetryItemProcessor implements ItemProcessor {private static final Logger LOGGER = LoggerFactory.getLogger(RetryItemProcessor.class); @Autowiredprivate CloseableHttpClient closeableHttpClient; @Overridepublic Transaction process(Transaction transaction) throws IOException, JSONException {LOGGER.info("Attempting to process user with id={}", transaction.getUserId()); HttpResponse response = fetchMoreUserDetails(transaction.getUserId()); //parse user's age and postCode from response and update transactionString result = EntityUtils.toString(response.getEntity()); JSONObject userObject = new JSONObject(result); transaction.setAge(Integer.parseInt(userObject.getString("age"))); transaction.setPostCode(userObject.getString("postCode")); return transaction; }private HttpResponse fetchMoreUserDetails(int id) throws IOException {final HttpGet request = new HttpGet("http://www.testapi.com:81/user/" + id); return closeableHttpClient.execute(request); }}

这里当然也可以使用RestTemplate进行调用，调用服务仅为了测试，读者可以搭建测试接口。

最终输出结果为

10000.02015-10-31 00:00:001234sammy10430222...

2. 给处理增加重试功能如果连接rest接口因为网络不稳定导致连接超时，那么批处理将失败。但这种错误并不是不能恢复，可以通过重试几次进行尝试。

因此我们配置批处理job在失败的情况下重试三次

@Configuration@EnableBatchProcessingpublic class SpringBatchRetryConfig {private static final String[] tokens = { "username", "userid", "transactiondate", "amount" }; private static final int TWO_SECONDS = 2000; @Autowiredprivate JobBuilderFactory jobBuilderFactory; @Autowiredprivate StepBuilderFactory stepBuilderFactory; @Value("input/recordRetry.csv")private Resource inputCsv; @Value("file:xml/retryOutput.xml")private Resource outputXml; public ItemReader itemReader(Resource inputData) throws ParseException {DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(); tokenizer.setNames(tokens); DefaultLineMapper lineMapper = new DefaultLineMapper<>(); lineMapper.setLineTokenizer(tokenizer); lineMapper.setFieldSetMapper(new RecordFieldSetMapper()); FlatFileItemReader reader = new FlatFileItemReader<>(); reader.setResource(inputData); reader.setLinesToSkip(1); reader.setLineMapper(lineMapper); return reader; }@Beanpublic CloseableHttpClient closeableHttpClient() {final RequestConfig config = RequestConfig.custom().setConnectTimeout(TWO_SECONDS).build(); return HttpClientBuilder.create().setDefaultRequestConfig(config).build(); }@Beanpublic ItemProcessor retryItemProcessor() {return new RetryItemProcessor(); }@Beanpublic ItemWriter itemWriter(Marshaller marshaller) {StaxEventItemWriter itemWriter = new StaxEventItemWriter<>(); itemWriter.setMarshaller(marshaller); itemWriter.setRootTagName("transactionRecord"); itemWriter.setResource(outputXml); return itemWriter; }@Beanpublic Marshaller marshaller() {Jaxb2Marshaller marshaller = new Jaxb2Marshaller(); marshaller.setClassesToBeBound(Transaction.class); return marshaller; }@Beanpublic Step retryStep(@Qualifier("retryItemProcessor") ItemProcessor processor,ItemWriter writer) throws ParseException {return stepBuilderFactory.get("retryStep").chunk(10).reader(itemReader(inputCsv)).processor(processor).writer(writer).faultTolerant().retryLimit(3).retry(ConnectTimeoutException.class).retry(DeadlockLoserDataAccessException.class).build(); }@Bean(name = "retryBatchJob")public Job retryJob(@Qualifier("retryStep") Step retryStep) {return jobBuilderFactory.get("retryBatchJob").start(retryStep).build(); }

这里调用faultTolerant()方法启用重试功能，并设置重试次数和对应异常。

3. 测试重试功能我们测试场景，期望接口在一定时间内返回年龄和邮编。前两次调用API抛出异常ConnectTimeoutException

第三次成功调用

@RunWith(SpringRunner.class)@SpringBatchTest@EnableAutoConfiguration@ContextConfiguration(classes = { SpringBatchRetryConfig.class })public class SpringBatchRetryIntegrationTest {private static final String TEST_OUTPUT = "xml/retryOutput.xml"; private static final String EXPECTED_OUTPUT = "src/test/resources/output/batchRetry/retryOutput.xml"; @Autowiredprivate JobLauncherTestUtils jobLauncherTestUtils; @MockBeanprivate CloseableHttpClient closeableHttpClient; @Mockprivate CloseableHttpResponse httpResponse; @Testpublic void whenEndpointAlwaysFail_thenJobFails() throws Exception {when(closeableHttpClient.execute(any())).thenThrow(new ConnectTimeoutException("Endpoint is down")); JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters()); JobInstance actualJobInstance = jobExecution.getJobInstance(); ExitStatus actualJobExitStatus = jobExecution.getExitStatus(); assertThat(actualJobInstance.getJobName(), is("retryBatchJob")); assertThat(actualJobExitStatus.getExitCode(), is("FAILED")); assertThat(actualJobExitStatus.getExitDescription(), containsString("org.apache.http.conn.ConnectTimeoutException")); }@Testpublic void whenEndpointFailsTwicePasses3rdTime_thenSuccess() throws Exception {FileSystemResource expectedResult = new FileSystemResource(EXPECTED_OUTPUT); FileSystemResource actualResult = new FileSystemResource(TEST_OUTPUT); //前两次调用失败，第三次继续执行when(httpResponse.getEntity()).thenReturn(new StringEntity("{ \"age\":10, \"postCode\":\"430222\" }")); when(closeableHttpClient.execute(any())).thenThrow(new ConnectTimeoutException("Timeout count 1")).thenThrow(new ConnectTimeoutException("Timeout count 2")).thenReturn(httpResponse); JobExecution jobExecution = jobLauncherTestUtils.launchJob(defaultJobParameters()); JobInstance actualJobInstance = jobExecution.getJobInstance(); ExitStatus actualJobExitStatus = jobExecution.getExitStatus(); assertThat(actualJobInstance.getJobName(), is("retryBatchJob")); assertThat(actualJobExitStatus.getExitCode(), is("COMPLETED")); AssertFile.assertFileEquals(expectedResult, actualResult); }private JobParameters defaultJobParameters() {JobParametersBuilder paramsBuilder = new JobParametersBuilder(); paramsBuilder.addString("jobID", String.valueOf(System.currentTimeMillis())); return paramsBuilder.toJobParameters(); }}

【基于Spring|基于Spring Batch 配置重试逻辑】
job成功执行
从日志可以看到两次失败，最终调用成功。

19:06:57.758 [main] INFO o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO o.b.batch.service.RetryItemProcessor - Attempting to process user with id=1234
19:06:57.758 [main] INFO o.b.batch.service.RetryItemProcessor - Attempting to process user with id=9999
19:06:57.773 [main] INFO o.s.batch.core.step.AbstractStep - Step: [retryStep] executed in 31ms

同时也定义了另一个测试，重试多次并失败，抛出异常 ConnectTimeoutException。

4. 总结本文我们学习如何配置Spring Batch的重试逻辑。通过示例学习配置并机型测试，仅为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。