想要在 Spring Batch 中为事务实现指数回退

Question

基本上就是标题。我正在使用 JDBC 项目读取器和 JDBC 项目写入器，并且我正在使用 API 通过处理器更改特定状态，如果 API 无法更改状态，我想使用指数退避来稍后重试。我不知道如何实现这个

Answer 1

您有两个选择：

1.在项目处理器中手动处理重试操作

基本上有两种方法可以做到这一点，无论是编程方式还是声明方式。

1.1 程序化方法

您首先根据需要定义带有退避策略的重试模板：

@Bean
public RetryTemplate retryTemplate() {
    // configure backoff policy
    ExponentialBackOffPolicy exponentialBackOffPolicy = new ExponentialBackOffPolicy();
    exponentialBackOffPolicy.setInitialInterval(1000);
    exponentialBackOffPolicy.setMultiplier(2.0);
    exponentialBackOffPolicy.setMaxInterval(10000);

    // configure retry policy
    SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
    simpleRetryPolicy.setMaxAttempts(5);

    // configure retry template
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(exponentialBackOffPolicy);
    retryTemplate.setRetryPolicy(simpleRetryPolicy);

    return retryTemplate;
}

然后在项目处理器中使用该重试模板：

import org.springframework.batch.item.ItemProcessor;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.support.RetryTemplate;

public class MyRetryableItemProcessor implements ItemProcessor {

    RetryTemplate retryTemplate;

    public MyRetryableItemProcessor(RetryTemplate retryTemplate) {
        this.retryTemplate = retryTemplate;
    }

    @Override
    public Object process(Object item) throws Exception {
        return retryTemplate.execute(new RetryCallback<Object, Exception>() {
            @Override
            public Object doWithRetry(RetryContext retryContext) throws Exception {
                // API call
                return item;
            }
        });
    }
}

1.2 使用注释的声明式方法

这是一个例子：

import org.springframework.batch.item.ItemProcessor;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

@Component
public class MyAnnotationBasedRetryableItemProcessor implements ItemProcessor {

    @Override
    @Retryable(backoff = @Backoff(delay = 1000L, maxDelay = 10000, multiplier = 2.0))
    public Object process(Object item) throws Exception {
        // Do API call
        return item;
    }

}

2.让 Spring Batch 通过使用容错步骤来为您处理重试

在这种情况下，您可以在容错步骤中设置自定义

RetryPolicy

：

@Bean
public Step step(StepBuilderFactory stepBuilderFactory) {
    // configure backoff policy
    ExponentialBackOffPolicy exponentialBackOffPolicy = new ExponentialBackOffPolicy();
    exponentialBackOffPolicy.setInitialInterval(1000);
    exponentialBackOffPolicy.setMultiplier(2.0);
    exponentialBackOffPolicy.setMaxInterval(10000);

    // configure retry policy
    SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
    simpleRetryPolicy.setMaxAttempts(5);
    
    return stepBuilderFactory.get("myStep")
            .<Integer, Integer>chunk(5)
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter())
            .faultTolerant()
            .retryPolicy(simpleRetryPolicy)
            .backOffPolicy(exponentialBackOffPolicy)
            .build();
}

请注意，在这种情况下，每当您的处理器抛出某个项目的异常时，就会逐项重试整个块（并且每个项目将在其自己的事务中重新处理）。

---

上面的示例使用了 spring-retry，因为您提到您对此有偏好。但同样的想法可以应用于任何其他容错库。