Spring Batch中如何在Job结束后统计所有Chunk的总处理记录数？

方案一：利用StepExecutionListener获取单Step处理统计 #

Spring Batch的StepExecution对象本身就包含了完整的处理统计数据，你可以自定义一个StepExecutionListener来获取这些数据，在Step执行完成后记录总处理数：

1. 自定义Step监听器：

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.stereotype.Component;

@Component
public class CustomStepStatsListener implements StepExecutionListener {

    @Override
    public void beforeStep(StepExecution stepExecution) {
        // 可选：Step执行前的初始化操作
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        // 获取核心统计数据
        long totalRead = stepExecution.getReadCount(); // 从数据库读取的总记录数
        long totalWritten = stepExecution.getWriteCount(); // 成功写入非关系库的记录数
        long totalSkipped = stepExecution.getSkipCount(); // 跳过的记录数（包含处理器/写入器抛出异常的）
        long totalProcessed = stepExecution.getProcessCount(); // 经过处理器处理的记录数（若处理器返回null则不纳入）

        // 这里根据你的需求记录统计，比如打印日志、存入数据库或缓存
        System.out.printf("Step执行完成：总读取%s条，总处理%s条，总写入%s条，跳过%s条%n", 
                          totalRead, totalProcessed, totalWritten, totalSkipped);
        
        // 如果需要把统计数据传递到Job级别，可以存入JobExecutionContext
        stepExecution.getJobExecution().getExecutionContext().putLong("totalWritten", totalWritten);
        
        return stepExecution.getExitStatus();
    }
}

2. 将监听器添加到你的Step配置中：
   修改你现有的Step Bean，加上这个监听器：

@Bean
public Step step1(JobRepository jobRepository,
                  PlatformTransactionManager transactionManager,
                  CustomStepStatsListener stepStatsListener) { // 注入自定义监听器
    return new StepBuilder("Stepname", jobRepository)
            .allowStartIfComplete(true)
            .<OriginEntity, DestinationEntity>chunk(chunkSize, transactionManager)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .faultTolerant()
            .skip(ProcessorException.class)
            .skip(WriterException.class)
            .skipLimit(skipLimit)
            .listener(itemReaderExecutionListener)
            .listener(itemProcessorExecutionListener)
            .listener(itemWriterExecutionListener)
            .listener(stepStatsListener) // 添加自定义Step统计监听器
            .build();
}

方案二：利用JobExecutionListener汇总多Step统计（如果是多Step任务） #

如果你的Job包含多个Step，需要在Job执行完成后汇总所有Step的处理数据，可以自定义JobExecutionListener：

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.stereotype.Component;

@Component
public class CustomJobStatsListener implements JobExecutionListener {

    @Override
    public void beforeJob(JobExecution jobExecution) {
        // 可选：Job执行前的初始化
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        long totalReadAllSteps = 0;
        long totalWrittenAllSteps = 0;

        // 遍历所有Step的执行统计
        for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
            totalReadAllSteps += stepExecution.getReadCount();
            totalWrittenAllSteps += stepExecution.getWriteCount();
        }

        // 汇总记录
        System.out.printf("Job执行完成：所有Step总读取%s条，总写入%s条%n", 
                          totalReadAllSteps, totalWrittenAllSteps);
    }
}

然后把这个监听器添加到Job配置中：

@Bean
public Job importJob(JobRepository jobRepository, Step step1, CustomJobStatsListener jobStatsListener) {
    return new JobBuilder("importJob", jobRepository)
            .start(step1)
            .listener(jobStatsListener) // 添加Job级统计监听器
            .build();
}

关键说明 #

• readCount：从Reader读取的总记录数，不管后续是否被过滤、跳过或写入失败
• writeCount：成功被Writer写入的记录数（对应Chunk提交成功的数量）
• skipCount：因为异常被跳过的记录数（你配置的ProcessorException和WriterException都会计入）
• processCount：经过Processor处理的记录数（如果Processor返回null，该记录不会被写入，也不会计入processCount）

你可以根据业务需求选择对应的统计值，比如如果需要统计"最终成功写入的总记录数"，就用writeCount；如果需要统计"从源库读取的总记录数"，就用readCount。

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