Spring Batch自定义ItemWriteListener场景下重试导致的计数统计异常问题求解

这个问题我之前也踩过坑，核心原因是Spring Batch的容错重试会多次触发ItemWriteListener的回调：当chunk首次失败时，onWriteError会把全量失败数更新到数据库；如果后续重试成功，afterWrite又会把全量成功数加进去，最终导致成功和失败计数都被错误累加。

核心思路 #

我们需要区分「临时失败（还会重试）」和「最终失败（重试耗尽）」：只有当chunk最终确定失败时，才更新失败计数；只要重试成功，就只统计成功计数，忽略之前的临时失败。

具体实现方案 #

利用Spring Retry的RetrySynchronizationManager获取当前重试上下文，判断当前失败是否已经达到重试上限，再决定是否更新数据库计数：

1. 修改自定义监听类

import org.springframework.batch.core.ItemWriteListener;
import org.springframework.retry.support.RetrySynchronizationManager;
import org.springframework.retry.RetryContext;

import java.util.List;

public class CustomItemWriterListener implements ItemWriteListener<Item> {

    private int retryLimit;

    // 通过构造器注入重试上限，和step配置保持一致
    public CustomItemWriterListener(int retryLimit) {
        this.retryLimit = retryLimit;
    }

    @Override
    public void afterWrite(List<? extends Item> items) {
        long failureCount = items.stream().filter(Item::hasErrors).count();
        long successCount = items.size() - failureCount;
        // 只有重试成功时才会走到这里，直接累加成功/失败数
        incrementCountsInDb(failureCount, successCount);
    }

    @Override
    public void onWriteError(Exception exception, List<? extends Item> items) {
        RetryContext retryContext = RetrySynchronizationManager.getContext();
        if (retryContext != null) {
            // 当前重试次数从0开始计数，比如retryLimit=3时，最多重试3次，最后一次失败时count=2
            int currentRetryCount = retryContext.getRetryCount();
            // 只有当重试次数达到上限时，才认为是最终失败，更新失败计数
            if (currentRetryCount >= retryLimit - 1) {
                int failureCount = items.size();
                incrementCountsInDb(failureCount, 0);
            }
            // 未达到重试上限的临时失败，不更新数据库，等待重试结果
        } else {
            // 没有重试上下文的情况（比如未开启重试），直接更新失败计数
            int failureCount = items.size();
            incrementCountsInDb(failureCount, 0);
        }
    }

    private void incrementCountsInDb(long failureCount, long successCount) {
        // 你的数据库更新逻辑
    }
}

2. 更新Step配置

在构建step时，把重试上限传入自定义监听：

stepBuilderFactory.get("testStep")
    .<String, Object>chunk(10)
    .reader(customReader())
    .processor(processor())
    .writer(customWriter())
    // 传入和retryLimit一致的数值
    .listener(new CustomItemWriterListener(3))
    .faultTolerant()
    .retry(DataAccessException.class)
    .retryLimit(3)
    .build();

其他可选方案 #

如果需要更严谨的统计（比如避免步骤崩溃导致的临时计数残留），可以结合StepExecutionContext跟踪每个chunk的状态：

• 在ChunkListener.beforeChunk中生成当前chunk的唯一标识，存入上下文
• onWriteError时，仅将失败记录关联到该chunk标识，不直接累加数据库计数
• afterWrite时，删除该chunk的失败记录，再累加成功计数
• 最后在StepExecutionListener.afterStep中清理未处理的chunk记录（针对步骤异常终止的情况）

这种方式实现稍复杂，但适合对统计准确性要求极高的场景。

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