Spring Batch FlatFileItemWriter数据块缺失问题排查求助

问题场景 #

数据库中存在12703条记录，但通过Spring Batch的RepositoryItemReader+FlatFileItemWriter导出后，输出文件仅写入7703条数据，恰好缺失一个chunk（chunk大小配置为5000）。添加ChunkListener监控后，未捕获到任何chunk失败日志，仅执行了2个chunk（预期为3个）。将ItemReader和ItemWriter的作用域从StepScope改为JobScope后，问题仍未解决。

相关组件代码 #

1. RepositoryItemReader实现 #

@Component
@StepScope
public class MyItemReader extends RepositoryItemReader<MyEntity> {

    public MyItemReader(MyEntityRepository repository) throws Exception {
        super();      

        this.setRepository(repository);
        this.setPageSize(1000);
        Map<String, Sort.Direction> sort = new HashMap<>();
        sort.put("id", Sort.Direction.ASC);
        this.setSort(sort);      
        this.setMethodName("findAll");
        this.afterPropertiesSet();
    }

}

2. ItemProcessor实现 #

@Component
@StepScope
public class MyItemProcessor implements ItemProcessor<MyEntity, String> {

    private final MyEntityRepository repository;

    public MyItemProcessor(MyEntityRepository repository) {
        this.repository = repository;
    }

    @Override
    public String process(MyEntity item) {
        // 业务逻辑：将实体转换为字符串行
        // 返回前更新文件生成状态
    }
}

3. FlatFileItemWriter实现 #

@Component
@StepScope
public class MyItemWriter extends FlatFileItemWriter<String> {
    public MyItemWriter(@Value("#{jobParameters['fileName']}") String fileName) throws Exception {
        this.setResource(new FileSystemResource(String.format("batch_files/%s", fileName)));
       this.setLineAggregator(new PassThroughLineAggregator<>());
       this.afterPropertiesSet();
    }
}

4. Step与Job配置 #

@Bean
public Step step(JobRepository jobRepository,
                 PlatformTransactionManager transactionManager,
                 MyItemReader myItemReader,
                 MyItemProcessor myItemProcessor,
                 MyItemWriter myItemWriter) {
    return new StepBuilder("fileGenerationStep", jobRepository)
            .<MyEntity, String>chunk(5000, transactionManager)               
            .reader(myItemReader)
            .processor(myItemProcessor)
            .writer(myItemWriter)
            .build();
}

@Bean
Job job(JobRepository jobRepository, Step step) {
    return new JobBuilder("fileJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(step)               
            .build();
}

5. ChunkListener实现 #

@Log4j2
public class MyChunkListener implements ChunkListener {
    private int counter;

    @Override
    public void afterChunk(ChunkContext context) {
        log.info("Completed chunks: {}, Chunk complete status: {}", counter, context.isComplete());
    }

    @Override
    public void beforeChunk(ChunkContext context) {
        log.info("Chunk started, Context:  {}", context.getStepContext().getStepExecutionContext());
        counter++;
    }

    @Override
    public void afterChunkError(ChunkContext context) {
        log.info("Chunk failed, Context:  {}", context.getStepContext().getStepExecutionContext());
        log.error("Chunk number " + counter + " failed");
    }
}

// 更新后的Step配置
@Bean
public Step step(JobRepository jobRepository,
                 PlatformTransactionManager transactionManager,
                 MyItemReader myItemReader,
                 MyItemProcessor myItemProcessor,
                 MyItemWriter myItemWriter) {
    return new StepBuilder("fileGenerationStep", jobRepository)
            .<MyEntity, String>chunk(5000, transactionManager)
            .listener(new MyChunkListener())
            .reader(myItemReader)
            .processor(myItemProcessor)
            .writer(myItemWriter)
            .build();
}

排查与解决方法 #

1. 检查RepositoryItemReader的分页逻辑 #

• 核心问题点：RepositoryItemReader的pageSize=1000小于chunk size=5000，意味着每个chunk需要读取5次分页数据。如果排序字段id存在重复（虽主键理论唯一，但需确认），或分页查询期间数据发生变动（比如processor中更新了记录），会导致分页结果偏移，遗漏数据。
• 解决：
  • 确保排序字段是唯一且不可变的（比如自增主键id），避免分页时结果集顺序变动。
  • 单独测试MyEntityRepository的findAll(Pageable)方法，验证分页结果是否与预期一致。

2. 排查ItemProcessor的过滤逻辑 #

• 核心问题点：如果ItemProcessor返回null，Spring Batch会过滤该条数据，不传递给Writer。若某个chunk的所有数据都被processor返回null，则该chunk不会写入文件，且不会触发chunk error。
• 解决：
  • 在processor中添加日志，统计每一条数据的处理结果（是否返回非null值）。
  • 查看Spring Batch元数据表BATCH_STEP_EXECUTION中的READ_COUNT、WRITE_COUNT、FILTER_COUNT字段：
    • 若FILTER_COUNT=5000，说明processor过滤了整整一个chunk的数据，需检查业务逻辑中是否存在批量返回null的情况。

3. 验证FlatFileItemWriter的写入逻辑 #

• 核心问题点：FlatFileItemWriter默认在chunk事务提交时写入文件，若事务隐性回滚（未抛出异常）或文件写入权限不足，可能导致chunk数据未写入。
• 解决：
  • 添加ItemWriteListener监控写入操作，统计每个chunk的写入数量。
  • 检查文件路径权限，确保应用有写入batch_files/目录的权限；验证FileSystemResource的路径是否正确，避免写入到错误路径。

4. 检查Spring Batch的Step执行统计 #

• 直接查看BATCH_STEP_EXECUTION表的核心字段：
  • READ_COUNT：实际读取的记录数，若为12703，说明reader正常读取了所有数据；若为10000，说明reader提前终止读取。
  • WRITE_COUNT：实际写入的记录数，若为7703，结合FILTER_COUNT可定位是过滤还是写入问题。
  • ROLLBACK_COUNT：若大于0，说明存在事务回滚，需排查回滚原因。

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