知識庫 / Spring / Spring Boot RSS 訂閱

1. 概述

默認情況下，Spring Batch 任務在執行過程中出現任何錯誤時都會失敗。然而，在某些情況下，我們可能希望提高應用程序的彈性，以應對間歇性故障。

在本快速教程中，我們將探索如何在 Spring Batch 框架中配置重試邏輯。

2. 示例用例

假設我們有一個批處理作業，它讀取一個輸入 CSV 文件：

username, userid, transaction_date, transaction_amount
sammy, 1234, 31/10/2015, 10000
john, 9999, 3/12/2015, 12321

然後，它通過調用 REST 端點，處理每個記錄，以獲取用户的 年齡 和 郵政編碼 屬性：

public class RetryItemProcessor implements ItemProcessor<Transaction, Transaction> {
    
    @Override
    public Transaction process(Transaction transaction) throws IOException {
        log.info("RetryItemProcessor, attempting to process: {}", transaction);
        HttpResponse response = fetchMoreUserDetails(transaction.getUserId());
        //parse user's age and postCode from response and update transaction
        ...
        return transaction;
    }
    ...
}

最後，它會生成一個彙總的輸出，格式為XML：

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

3. 為 ItemProcessor 添加重試機制

現在，如果由於網絡延遲導致與 REST 端點的連接超時，將會發生什麼？如果發生這種情況，我們的批次作業將會失敗。

在這樣的情況下，我們更傾向於在發生失敗時重試處理失敗的項目幾次。因此，讓我們配置批次作業，在發生失敗時最多執行三次重試：

@Bean
public Step retryStep(
  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();
}

在這裏，我們調用了faultTolerant()以啓用重試功能。此外，我們使用retry和retryLimit來定義符合重試條件的異常以及每個項目的最大重試次數，分別。

4. 測試重試

讓我們設計一個測試場景，其中返回 age 和 postCode 的 REST 端點僅短暫不可用。 在此測試場景中，我們只會針對前兩個 API 調用獲得 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\" }"));
 
    //fails for first two calls and passes third time onwards
    when(httpClient.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);
}

在這裏，我們的工作已成功完成。此外，日誌也顯示第一個記錄，id=1234，在第二次嘗試後失敗，最終在第三次重試中成功。

19:06:57.742 [main] INFO  o.s.batch.core.job.SimpleStepHandler - Executing step: [retryStep]
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

同樣，我們來做一個測試用例，看看當所有重試都耗盡時會發生什麼：

@Test
public void whenEndpointAlwaysFail_thenJobFails() throws Exception {
    when(httpClient.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"));
}

在這種情況下，對於第一個記錄，嘗試了三次重試，最終由於發生ConnectTimeoutException錯誤而失敗。

5. 使用 XML 配置重試

最後，讓我們看一下上述配置的 XML 對應內容:

<batch:job id="retryBatchJob">
    <batch:step id="retryStep">
        <batch:tasklet>
            <batch:chunk reader="itemReader" writer="itemWriter"
              processor="retryItemProcessor" commit-interval="10"
              retry-limit="3">
                <batch:retryable-exception-classes>
                    <batch:include class="org.apache.http.conn.ConnectTimeoutException"/>
                    <batch:include class="org.springframework.dao.DeadlockLoserDataAccessException"/>
                </batch:retryable-exception-classes>
            </batch:chunk>
        </batch:tasklet>
    </batch:step>
</batch:job>

6. 結論

在本文中，我們學習瞭如何在 Spring Batch 中配置重試邏輯。我們考察了 Java 和 XML 配置兩種方式。

我們還使用單元測試來觀察重試在實踐中的效果。