Spring Batch中如何在ChunkListener内获取已处理条目数量

核心实现逻辑为利用同Chunk生命周期内上下文共享的特性，先在ItemWriteListener中暂存当前Chunk处理的条目数，再在事务提交后触发的ChunkListener中读取数值并持久化，具体实现步骤如下：

1. 实现ItemWriteListener暂存处理条数 #

在afterWrite方法中获取当前Chunk的上下文对象，将本次处理的条目数存入上下文属性：

import org.springframework.batch.core.scope.context.ThreadLocalChunkContextHolder;
import org.springframework.batch.core.listener.ItemListenerSupport;

public class CountItemWriteListener<S> extends ItemListenerSupport<S, S> {
    @Override
    public void afterWrite(List<? extends S> items) {
        // 获取当前Chunk的上下文
        var chunkContext = ThreadLocalChunkContextHolder.getChunkContext();
        // 存储本次处理的条目数
        chunkContext.setAttribute("currentChunkProcessedCount", items.size());
    }
}

2. 实现ChunkListener读取计数并持久化 #

在事务提交后触发的afterChunk方法中，读取之前存入上下文的条目数，执行数据库保存操作：

import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.scope.context.ChunkContext;

public class CountChunkListener implements ChunkListener {
    @Override
    public void afterChunk(ChunkContext context) {
        Integer processedCount = (Integer) context.getAttribute("currentChunkProcessedCount");
        // 仅当存在有效计数时执行保存
        if (processedCount != null && processedCount > 0) {
            saveToDatabase(processedCount);
            // 可选：清理上下文属性，避免异常场景下重复统计
            context.removeAttribute("currentChunkProcessedCount");
        }
    }

    @Override
    public void beforeChunk(ChunkContext context) {
        // 无需实现逻辑
    }

    @Override
    public void afterChunkError(ChunkContext context) {
        // 可选：chunk执行出错时清理上下文属性
        context.removeAttribute("currentChunkProcessedCount");
    }

    private void saveToDatabase(int count) {
        // 你的持久化逻辑实现
    }
}

3. 监听器注册 #

将上述两个监听器注册到对应的Step中即可生效，示例配置：

@Bean
public Step yourStep(ItemReader<YourEntity> reader,
                     ItemProcessor<YourEntity, YourEntity> processor,
                     ItemWriter<YourEntity> writer) {
    return stepBuilderFactory.get("yourStep")
            .<YourEntity, YourEntity>chunk(100)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .listener(new CountItemWriteListener<>())
            .listener(new CountChunkListener())
            .build();
}

方案说明 #

• 上下文隔离：ChunkContext是Chunk级线程隔离的，多线程并行执行场景下也不会出现计数串用问题
• 异常兼容：如果Chunk执行回滚，afterWrite方法不会被触发，上下文不会存入无效计数，不会出现统计错误
• 执行时序完全符合需求：计数存储逻辑在Chunk事务提交后执行，不会出现主业务回滚但计数已保存的不一致问题

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