Spring Batch异步处理器/写入器重试与终止问题咨询

一、异步组件下的重试+任务终止配置示例 #

首先得说清楚为啥普通的Spring Batch步骤重试在AsyncProcessor/AsyncItemWriter下失效——因为异步组件是把任务扔到独立线程池里跑的，原来的Step重试逻辑是在主线程，根本感知不到异步线程里的异常。所以咱们得换个思路：把重试逻辑放到异步组件的委托处理器里，用Spring Retry来控制重试次数，再配置AsyncItemWriter把异常抛回主线程触发任务终止。

话不多说，直接上代码：

1. 带重试逻辑的委托Processor #

这个Processor是实际调用第三方API的地方，用RetryTemplate封装重试逻辑：

@Component
public class ThirdPartyApiProcessor implements ItemProcessor<InputItem, OutputItem> {

    private final RestTemplate restTemplate;
    private final RetryTemplate retryTemplate;

    // 构造注入依赖
    public ThirdPartyApiProcessor(RestTemplate restTemplate, RetryTemplate retryTemplate) {
        this.restTemplate = restTemplate;
        this.retryTemplate = retryTemplate;
    }

    @Override
    public OutputItem process(InputItem item) throws Exception {
        // 用RetryTemplate包裹API调用，达到重试上限后自动抛出ExhaustedRetryException
        return retryTemplate.execute(retryContext -> {
            ResponseEntity<ApiResponse> apiResponse = restTemplate.postForEntity(
                    "https://third-party-api.com/process",
                    item.getPayload(),
                    ApiResponse.class
            );

            // 非2xx响应直接抛出异常触发重试
            if (!apiResponse.getStatusCode().is2xxSuccessful()) {
                throw new ApiUnavailableException("第三方API调用失败，状态码：" + apiResponse.getStatusCode());
            }

            // 把API结果映射成Batch需要的OutputItem
            OutputItem output = new OutputItem();
            output.setId(item.getId());
            output.setResult(apiResponse.getBody().getData());
            return output;
        });
    }
}

2. 配置RetryTemplate #

这里定义重试的规则：最多重试3次，只针对API相关异常，每次重试间隔1秒：

@Configuration
public class RetryConfig {

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        // 重试策略：最多3次尝试（含第一次调用），指定哪些异常需要重试
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
        retryableExceptions.put(ApiUnavailableException.class, true);
        retryableExceptions.put(RestClientException.class, true); // 覆盖HTTP客户端的所有异常
        retryPolicy.setRetryableExceptions(retryableExceptions);

        // 退避策略：每次重试间隔1秒，避免频繁打API
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000);

        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        return retryTemplate;
    }
}

3. 配置AsyncProcessor、AsyncItemWriter和Job Step #

关键是给AsyncItemWriter设置throwsExceptionOnFailure(true)，这样异步线程里的异常会被抛回主线程，触发任务终止：

@Configuration
@EnableBatchProcessing
public class AsyncBatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final ThirdPartyApiProcessor delegateProcessor;
    private final ItemWriter<OutputItem> delegateWriter;

    public AsyncBatchConfig(JobBuilderFactory jobBuilderFactory,
                            StepBuilderFactory stepBuilderFactory,
                            ThirdPartyApiProcessor delegateProcessor,
                            ItemWriter<OutputItem> delegateWriter) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.delegateProcessor = delegateProcessor;
        this.delegateWriter = delegateWriter;
    }

    @Bean
    public ItemProcessor<InputItem, Future<OutputItem>> asyncProcessor() {
        AsyncItemProcessor<InputItem, OutputItem> asyncProcessor = new AsyncItemProcessor<>();
        asyncProcessor.setDelegate(delegateProcessor);
        // 配置你需要的线程池，这里是5个线程
        asyncProcessor.setTaskExecutor(asyncTaskExecutor());
        return asyncProcessor;
    }

    @Bean
    public ItemWriter<Future<OutputItem>> asyncItemWriter() {
        AsyncItemWriter<OutputItem> asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(delegateWriter);
        // 核心配置：让异步任务的异常抛回主线程，触发任务终止
        asyncItemWriter.setThrowsExceptionOnFailure(true);
        return asyncItemWriter;
    }

    @Bean
    public TaskExecutor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(10);
        executor.setThreadNamePrefix("batch-async-thread-");
        executor.initialize();
        return executor;
    }

    @Bean
    public Step asyncProcessingStep() {
        return stepBuilderFactory.get("asyncProcessingStep")
                .<InputItem, Future<OutputItem>>chunk(100)
                .reader(itemReader()) // 替换成你自己的ItemReader
                .processor(asyncProcessor())
                .writer(asyncItemWriter())
                // 配置Step的容错：不跳过任何失败项，遇到重试耗尽的异常直接终止任务
                .faultTolerant()
                .skipLimit(0)
                .noSkip(ExhaustedRetryException.class)
                .noSkip(ApiUnavailableException.class)
                .build();
    }

    @Bean
    public Job asyncProcessingJob() {
        return jobBuilderFactory.get("asyncProcessingJob")
                .start(asyncProcessingStep())
                .build();
    }

    // 示例ItemReader，实际项目中替换成你的数据源读取逻辑
    @Bean
    public ItemReader<InputItem> itemReader() {
        List<InputItem> testItems = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            InputItem item = new InputItem();
            item.setId(i);
            item.setPayload("test-payload-" + i);
            testItems.add(item);
        }
        return new ListItemReader<>(testItems);
    }
}

二、关于部分记录重试次数不一致的问题 #

你说设置了重试3次，但有的记录重试2次、有的3次，最后任务因ExhaustedRetryException失败——这完全是预期行为，原因有这几点：

1. 异步并行执行：每个记录的处理在独立线程里，重试的时间线不一样。比如A记录第一次调用就失败，开始重试；B记录可能前两次都成功，第三次才失败，重试次数自然不同。
2. 重试次数的计数方式：Spring Retry的setMaxAttempts(3)是指包括第一次尝试在内的总调用次数。比如重试2次意味着：1次正常调用 + 2次重试 = 总共3次，刚好达到阈值；如果你看到某条记录重试了3次，那可能是日志计数的问题（比如有的日志把第一次尝试也算成“重试”）。
3. 任务终止的时机：只要有一个异步任务触发了重试耗尽异常，AsyncItemWriter就会立刻把异常抛回主线程，任务直接终止。这时候其他正在重试的任务可能还没达到上限，所以最终会看到不同的重试次数。

三、Spring Retry vs Spring Batch 步骤容错重试的核心区别 #

这俩经常被搞混，我给你梳理下核心差异：

• 作用范围：Spring Retry是通用框架，能给任何Java方法加重试；而Spring Batch的步骤重试是专门为Batch的Step/Chunk流程设计的，只能用在Batch场景里。
• 异步兼容性：Spring Retry支持同步和异步场景，因为它是在方法调用层级处理重试；而Batch的步骤重试依赖主线程控制，在AsyncProcessor/AsyncItemWriter这种异步场景下，主线程根本拿不到异步线程的异常，所以完全失效。
• 重试粒度：Spring Retry可以精确到单个Item的处理；Batch步骤重试默认是重试整个Chunk（如果Chunk里有一个Item失败，整个Chunk都会重试），虽然也能配置成单个Item重试，但只在同步场景下有效。
• 集成方式：Batch步骤重试只需要在Step配置里加faultTolerant()就能启用，非常简单；Spring Retry需要手动把RetryTemplate注入到你的Processor/Writer里，自定义重试逻辑。

简单说：同步Batch任务用自带的步骤重试就行；异步组件必须用Spring Retry在委托组件里处理重试，不然根本控制不了。

内容的提问来源于stack exchange，提问作者ravicandy1234