Spring Batch：自定义统计次数并从Processor感知Job完成的问题

首先得明确：Processor本身是面向单个Item处理的组件，它无法直接感知整个Job的完成状态。不过我们可以通过「共享统计数据+Job监听器」的组合来实现你的需求——在Processor中记录第三方调用的成功/失败次数，等Job完全结束后统一写入日志。

下面分两种常见场景给出具体实现方案：

场景1：单线程/单分区Step（简单场景） #

这种情况下，我们可以用一个全局共享的统计Bean来存储计数，然后通过JobExecutionListener在Job完成后读取这些数据并输出日志。

1. 创建线程安全的统计Bean #

这个Bean用来存储全局的成功/失败次数，因为Processor可能在多线程环境下运行（比如TaskExecutorStep），所以要用线程安全的类（比如AtomicInteger）：

@Component
public class ThirdPartyCallStats {
    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger failureCount = new AtomicInteger(0);

    public void incrementSuccess() {
        successCount.incrementAndGet();
    }

    public void incrementFailure() {
        failureCount.incrementAndGet();
    }

    public int getSuccessCount() {
        return successCount.get();
    }

    public int getFailureCount() {
        return failureCount.get();
    }

    // 可选：如果Job可能重复执行，添加重置方法
    public void reset() {
        successCount.set(0);
        failureCount.set(0);
    }
}

2. 在Processor中注入统计Bean并更新计数 #

在你的Processor里，每次调用第三方后更新统计值：

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

    private final ThirdPartyCallStats stats;
    private final ThirdPartyService thirdPartyService;

    // 构造注入（推荐Spring 4.x+的方式）
    public MyItemProcessor(ThirdPartyCallStats stats, ThirdPartyService thirdPartyService) {
        this.stats = stats;
        this.thirdPartyService = thirdPartyService;
    }

    @Override
    public OutputItem process(InputItem item) throws Exception {
        try {
            ThirdPartyResponse response = thirdPartyService.call(item);
            if (response.isSuccess()) {
                stats.incrementSuccess();
                return mapToOutputItem(item, response); // 映射为输出Item
            } else {
                stats.incrementFailure();
                return null; // 返回null会跳过写入Writer，根据你的需求调整
            }
        } catch (Exception e) {
            stats.incrementFailure();
            // 可以选择抛出异常让Step处理，或者返回null
            throw new ItemProcessingException("调用第三方失败，Item ID: " + item.getId(), e);
        }
    }

    // 自定义映射方法，根据你的业务实现
    private OutputItem mapToOutputItem(InputItem item, ThirdPartyResponse response) {
        OutputItem output = new OutputItem();
        output.setId(item.getId());
        output.setResult(response.getResult());
        return output;
    }
}

3. 创建Job完成监听器并写入日志 #

实现JobExecutionListener接口，在afterJob方法中读取统计数据并写入日志：

@Component
public class JobCompletionStatsListener implements JobExecutionListener {

    private static final Logger logger = LoggerFactory.getLogger(JobCompletionStatsListener.class);
    private final ThirdPartyCallStats stats;

    public JobCompletionStatsListener(ThirdPartyCallStats stats) {
        this.stats = stats;
    }

    @Override
    public void beforeJob(JobExecution jobExecution) {
        // 可选：Job开始时重置统计值（如果Job可能重复执行）
        stats.reset();
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        // Job完成后输出统计日志
        logger.info("第三方调用统计结果：成功 {} 次，失败 {} 次", 
            stats.getSuccessCount(), stats.getFailureCount());
        // 如果需要写入特定日志文件，可以通过日志框架配置单独的Appender，比手动写FileWriter更可靠
    }
}

4. 在Job配置中注册监听器 #

把监听器绑定到你的Job上：

@Configuration
@EnableBatchProcessing
public class BatchJobConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final MyItemProcessor itemProcessor;
    private final JobCompletionStatsListener statsListener;

    public BatchJobConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                          MyItemProcessor itemProcessor, JobCompletionStatsListener statsListener) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.itemProcessor = itemProcessor;
        this.statsListener = statsListener;
    }

    @Bean
    public Step myStep(ItemReader<InputItem> itemReader, ItemWriter<OutputItem> itemWriter) {
        return stepBuilderFactory.get("myStep")
                .<InputItem, OutputItem>chunk(10)
                .reader(itemReader)
                .processor(itemProcessor)
                .writer(itemWriter)
                .build();
    }

    @Bean
    public Job myJob(Step myStep) {
        return jobBuilderFactory.get("myJob")
                .incrementer(new RunIdIncrementer())
                .listener(statsListener) // 注册Job监听器
                .flow(myStep)
                .end()
                .build();
    }
}

场景2：分区/多线程Step（复杂场景） #

如果你的Step是分区的（PartitionedStep）或者用了多线程TaskExecutor，每个Worker/线程会有自己的Processor实例，这时候用全局Bean虽然也能工作，但更推荐用ExecutionContext来存储每个Step的统计值，最后在Job层面汇总。

1. 修改Processor实现StepExecutionListener #

让Processor实现StepExecutionListener，在beforeStep中获取当前Step的ExecutionContext，然后在process方法中更新计数：

@Component
@StepScope // 分区场景下推荐用StepScope
public class MyItemProcessor implements ItemProcessor<InputItem, OutputItem>, StepExecutionListener {

    private StepExecution stepExecution;
    private final ThirdPartyService thirdPartyService;

    public MyItemProcessor(ThirdPartyService thirdPartyService) {
        this.thirdPartyService = thirdPartyService;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
        // 初始化当前Step的统计值
        stepExecution.getExecutionContext().putInt("successCount", 0);
        stepExecution.getExecutionContext().putInt("failureCount", 0);
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }

    @Override
    public OutputItem process(InputItem item) throws Exception {
        try {
            ThirdPartyResponse response = thirdPartyService.call(item);
            if (response.isSuccess()) {
                // 原子更新ExecutionContext中的计数
                int currentSuccess = stepExecution.getExecutionContext().getInt("successCount");
                stepExecution.getExecutionContext().putInt("successCount", currentSuccess + 1);
                return mapToOutputItem(item, response);
            } else {
                int currentFailure = stepExecution.getExecutionContext().getInt("failureCount");
                stepExecution.getExecutionContext().putInt("failureCount", currentFailure + 1);
                return null;
            }
        } catch (Exception e) {
            int currentFailure = stepExecution.getExecutionContext().getInt("failureCount");
            stepExecution.getExecutionContext().putInt("failureCount", currentFailure + 1);
            throw new ItemProcessingException("调用第三方失败，Item ID: " + item.getId(), e);
        }
    }

    // 映射方法...
}

2. 修改Job监听器汇总统计 #

在Job监听器中遍历所有StepExecution，汇总每个Step的统计值：

@Component
public class JobCompletionStatsListener implements JobExecutionListener {

    private static final Logger logger = LoggerFactory.getLogger(JobCompletionStatsListener.class);

    @Override
    public void afterJob(JobExecution jobExecution) {
        int totalSuccess = 0;
        int totalFailure = 0;

        // 遍历所有StepExecution，汇总统计数据
        for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
            totalSuccess += stepExecution.getExecutionContext().getInt("successCount", 0);
            totalFailure += stepExecution.getExecutionContext().getInt("failureCount", 0);
        }

        logger.info("第三方调用总统计：成功 {} 次，失败 {} 次", totalSuccess, totalFailure);
    }
}

关键注意事项 #

• 线程安全：如果用全局统计Bean，必须确保它是线程安全的（比如用AtomicInteger、ConcurrentHashMap）；
• Job重复执行：如果Job可能被重复触发，记得在beforeJob中重置统计值，避免累计之前的计数；
• 日志文件配置：如果需要把统计结果写入特定日志文件，建议通过Logback/Log4j2等日志框架配置单独的Appender，不要手动用FileWriter，这样更可靠且便于维护。

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