快捷導(dǎo)航

基于Spring Batch 配置重試邏輯

更新時(shí)間：2021年09月08日 08:39:15 作者：夢想畫家

這篇文章主要介紹了Spring Batch 配置重試邏輯，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

1. 應(yīng)用示例

批處理應(yīng)用讀取csv文件

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

然后，通過調(diào)用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<Transaction, Transaction> {
    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://www.testapi.com:81/user/" + id);
        return closeableHttpClient.execute(request);
    }
}

這里當(dāng)然也可以使用RestTemplate進(jìn)行調(diào)用，調(diào)用服務(wù)僅為了測試，讀者可以搭建測試接口。

最終輸出結(jié)果為

<transactionRecord>
    <transactionRecord>
        <amount>10000.0</amount>
        <transactionDate>2015-10-31 00:00:00</transactionDate>
        <userId>1234</userId>
        <username>sammy</username>
        <age>10</age>
        <postCode>430222</postCode>
    </transactionRecord>
    ...
</transactionRecord>

2. 給處理增加重試功能

如果連接rest接口因?yàn)榫W(wǎng)絡(luò)不穩(wěn)定導(dǎo)致連接超時(shí)，那么批處理將失敗。但這種錯(cuò)誤并不是不能恢復(fù)，可以通過重試幾次進(jìn)行嘗試。

因此我們配置批處理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<Transaction> itemReader(Resource inputData) throws ParseException {
        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
        tokenizer.setNames(tokens);
        DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<>();
        lineMapper.setLineTokenizer(tokenizer);
        lineMapper.setFieldSetMapper(new RecordFieldSetMapper());
        FlatFileItemReader<Transaction> 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<Transaction, Transaction> retryItemProcessor() {
        return new RetryItemProcessor();
    }
    @Bean
    public ItemWriter<Transaction> itemWriter(Marshaller marshaller) {
        StaxEventItemWriter<Transaction> 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 retryStep(@Qualifier("retryItemProcessor") ItemProcessor<Transaction, Transaction> processor,
      ItemWriter<Transaction> writer) throws ParseException {
        return stepBuilderFactory.get("retryStep")
          .<Transaction, Transaction>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();
    }

這里調(diào)用faultTolerant()方法啟用重試功能，并設(shè)置重試次數(shù)和對(duì)應(yīng)異常。

3. 測試重試功能

我們測試場景，期望接口在一定時(shí)間內(nèi)返回年齡和郵編。前兩次調(diào)用API拋出異常ConnectTimeoutException

第三次成功調(diào)用

@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);
        //前兩次調(diào)用失敗，第三次繼續(xù)執(zhí)行
        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成功執(zhí)行

從日志可以看到兩次失敗，最終調(diào)用成功。

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

同時(shí)也定義了另一個(gè)測試，重試多次并失敗，拋出異常 ConnectTimeoutException。