Spring Batch 5.x如何按bookingId分组记录？基于RepositoryItemReader实现

1. 核心思路 #

Spring Batch 5支持分组处理，通过自定义ItemStreamReader包装RepositoryItemReader实现按bookingId分组，再结合ItemGroupProcessor和ItemGroupWriter完成组级别的业务处理与结果写入。

2. 具体实现步骤 #

2.1 定义业务实体与Repository #

假设业务实体为BookingRecord，先完成基础定义：

// 业务记录实体
public class BookingRecord {
    private Long bookingId;
    private String detailData;
    // getter、setter、构造方法
}

// JPA Repository
public interface BookingRecordRepository extends JpaRepository<BookingRecord, Long> {
    // 按bookingId升序查询，确保同组记录连续返回
    List<BookingRecord> findAllByOrderByBookingIdAsc();
}

2.2 自定义分组ItemReader #

包装RepositoryItemReader，将单条记录按bookingId聚合为分组列表：

@Component
public class GroupedBookingItemReader implements ItemStreamReader<List<BookingRecord>> {

    private final RepositoryItemReader<BookingRecord> delegateReader;
    private BookingRecord currentRecord;

    public GroupedBookingItemReader(BookingRecordRepository repository) {
        this.delegateReader = new RepositoryItemReader<>();
        delegateReader.setRepository(repository);
        delegateReader.setMethodName("findAllByOrderByBookingIdAsc");
        delegateReader.setPageSize(100); // 根据数据量调整分页大小
        // 配置排序规则，保证同bookingId记录连续
        Map<String, Sort.Direction> sortMap = new HashMap<>();
        sortMap.put("bookingId", Sort.Direction.ASC);
        delegateReader.setSort(sortMap);
    }

    @Override
    public List<BookingRecord> read() throws Exception {
        if (currentRecord == null) {
            currentRecord = delegateReader.read();
            if (currentRecord == null) {
                return null; // 无更多记录
            }
        }

        Long targetBookingId = currentRecord.getBookingId();
        List<BookingRecord> group = new ArrayList<>();
        group.add(currentRecord);

        // 读取后续同bookingId的记录
        BookingRecord nextRecord;
        while ((nextRecord = delegateReader.read()) != null) {
            if (nextRecord.getBookingId().equals(targetBookingId)) {
                group.add(nextRecord);
            } else {
                currentRecord = nextRecord;
                return group;
            }
        }

        currentRecord = null; // 最后一组读取完成，重置缓存
        return group;
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        delegateReader.open(executionContext);
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        delegateReader.update(executionContext);
    }

    @Override
    public void close() throws ItemStreamException {
        delegateReader.close();
    }
}

2.3 实现分组处理器(ItemGroupProcessor) #

创建针对整个分组的业务处理器，输入为分组记录列表，输出为处理后的结果：

// 处理结果实体
public class BookingResult {
    private Long bookingId;
    private int recordCount;
    private String processSummary;
    // getter、setter、构造方法
}

// 分组业务处理器
@Component
public class BookingGroupProcessor implements ItemGroupProcessor<List<BookingRecord>, BookingResult> {

    @Override
    public BookingResult processGroup(List<BookingRecord> group) throws Exception {
        Long bookingId = group.get(0).getBookingId();
        int count = group.size();
        // 此处添加自定义业务逻辑，比如数据汇总、校验、转换等
        String summary = "完成bookingId=" + bookingId + "的" + count + "条记录处理";
        return new BookingResult(bookingId, count, summary);
    }
}

2.4 实现分组写入器(ItemGroupWriter) #

创建按组写入结果的组件，支持批量或单组写入：

@Component
public class BookingGroupWriter implements ItemGroupWriter<BookingResult> {

    private final BookingResultRepository resultRepository; // 注入结果存储Repository

    public BookingGroupWriter(BookingResultRepository resultRepository) {
        this.resultRepository = resultRepository;
    }

    @Override
    public void write(List<? extends BookingResult> results) throws Exception {
        // 批量写入分组结果
        resultRepository.saveAll(results);
        // 如需单组单独处理，可遍历results执行逻辑
        for (BookingResult result : results) {
            System.out.println("已写入结果：" + result.getProcessSummary());
        }
    }
}

2.5 配置Batch Job #

组装Job与Step，整合上述组件：

@Configuration
@EnableBatchProcessing
public class BookingBatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final GroupedBookingItemReader groupedBookingItemReader;
    private final BookingGroupProcessor bookingGroupProcessor;
    private final BookingGroupWriter bookingGroupWriter;

    public BookingBatchConfig(JobBuilderFactory jobBuilderFactory,
                              StepBuilderFactory stepBuilderFactory,
                              GroupedBookingItemReader groupedBookingItemReader,
                              BookingGroupProcessor bookingGroupProcessor,
                              BookingGroupWriter bookingGroupWriter) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.groupedBookingItemReader = groupedBookingItemReader;
        this.bookingGroupProcessor = bookingGroupProcessor;
        this.bookingGroupWriter = bookingGroupWriter;
    }

    @Bean
    public Step bookingProcessStep() {
        return stepBuilderFactory.get("bookingProcessStep")
                .<List<BookingRecord>, BookingResult>chunk(5) // chunk大小为单次处理的分组数量
                .reader(groupedBookingItemReader)
                .processor(bookingGroupProcessor)
                .writer(bookingGroupWriter)
                .build();
    }

    @Bean
    public Job bookingProcessJob() {
        return jobBuilderFactory.get("bookingProcessJob")
                .start(bookingProcessStep())
                .build();
    }
}

3. 关键注意事项 #

• 排序必须保证同组连续：Repository查询必须按bookingId排序，否则自定义Reader无法正确聚合同组记录（若同组记录被拆分到不同分页，会导致分组不完整）。
• Chunk参数含义：此处的chunk指单次处理的分组数量，而非单条记录数，需根据业务压力调整。
• ItemStream接口实现：自定义Reader必须实现ItemStreamReader，确保Batch的重启、恢复功能正常，避免数据重复处理或丢失。

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