基于Spring Batch 配置重试逻辑

基于Spring Batch 配置重试逻辑

目录1. 应用示例批处理应用读取csv文件处理类如下最终输出结果为2. 给处理增加重试功能因此我们配置批处理job在失败的情况下重试三次3. 测试重试功能第三次成功调用job成功执行4. 总结

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

1. 应用示例

批处理应用读取csv文件

sammy, 1234, 31/10/2015, 10000

john, 9999, 3/12/2015, 12321

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

@XmlRootElement(name = “transactionRecord”)

@Data

public 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);

@Autowired

private CloseableHttpClient closeableHttpClient;

@Override

public 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 transaction

String 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://testapi.com:81/user/" + id);

return closeableHttpClient.execute(request);

}

}

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

最终输出结果为

10000.0

2015-10-31 00:00:00

1234

sammy

10

430222

...

2. 给处理增加重试功能

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

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

@Configuration

@EnableBatchProcessing

public class SpringBatchRetryConfig {

private static final String[] tokens = { "username", "userid", "transactiondate", "amount" };

private static final int TWO_SECONDS = 2000;

@Autowired

private JobBuilderFactory jobBuilderFactory;

@Autowired

private 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 = IcQkWNnew DefaultLineMapper<>();

lineMapper.setLineTokenizer(tokenizer);

lineMapper.setFieldSetMapper(new RecordFieldSetMapper());

FlatFileItemReader reader = new FlatFileItemReader<>();

reader.setResource(inputData);

reader.setLinesToSkip(1);

reader.setLineMapper(lineMapper);

return reader;

}

@Bean

public CloseableHttpClient closeableHttpClient() {

final RequestConfig config = RequestConfig.custom()

.setConnectTimeout(TWO_SECONDS)

.build();

return HttpClientBuilder.create().setDefaultRequestConfig(config).build();

}

@Bean

public ItemProcessor retryItemProcessor() {

return new RetryItemProcessor();

}

@Bean

public ItemWriter itemWriter(Marshaller marshaller) {

StaxEventItemWriter itemWriter = new StaxEventItemWriter<>();

itemWriter.setMarshaller(marshaller);

itemWriter.setRootTagName("transactionRecord");

itemWriter.setResource(outputXml);

return itemWriter;

}

@Bean

public Marshaller marshaller() {

Jaxb2Marshaller marshaller = new Jaxb2Marshaller();

marshaller.setClassesToBeBound(Transaction.class);

return marshaller;

}

@Bean

public Step retryStephttp://(@Qualifier("retryItemProcessor") ItemProcessor processor,

ItemWriter writer) throws ParseException {

return stepBuilderFactory.get("retryStep")

.chunk(10)

http:// .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";

@Autowired

private JobLauncherTestUtils jobLauncherTestUtils;

@MockBean

private CloseableHttpClient closeableHttpClient;

@Mock

private CloseableHttpResponse httpResponse;

@Test

public 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"));

}

@Test

public 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();

}

}

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.RetryIhttp://temProcessor - 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. 总结

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。