如何使用Spring Batch从PostgreSQL数据库获取数据列表

当然可行，Spring Batch原生支持批量读取数据并将列表传递给处理器，尤其在PostgreSQL场景下可以通过两种主流方式实现：

方式一：基于Chunk模型的批量处理（推荐） #

Spring Batch的核心Chunk处理模型本身就为批量操作设计，通过配置commit-interval控制单次处理的数据量，配合PostgreSQL的批量读取参数，就能实现把整批数据列表传给处理器。

1. 配置批量读取的ItemReader #

使用JdbcCursorItemReader（适合中小数据量）或JdbcPagingItemReader（适合大数据量分页），通过fetchSize参数设置PostgreSQL每次从数据库拉取的行数，确保数据批量加载：

@Bean
public JdbcCursorItemReader<YourEntity> batchItemReader(DataSource dataSource) {
    return new JdbcCursorItemReaderBuilder<YourEntity>()
            .dataSource(dataSource)
            .sql("SELECT id, name, value FROM your_table WHERE status = ?")
            .parameterValues("PENDING")
            .rowMapper((rs, rowNum) -> {
                YourEntity entity = new YourEntity();
                entity.setId(rs.getLong("id"));
                entity.setName(rs.getString("name"));
                entity.setValue(rs.getInt("value"));
                return entity;
            })
            .fetchSize(100) // 每次从PostgreSQL批量获取100条数据
            .build();
}

2. 实现批量处理器 #

不要用默认的单条数据处理器，直接实现接收列表的ItemProcessor，一次性处理整批数据：

@Component
public class BatchYourEntityProcessor implements ItemProcessor<List<YourEntity>, List<ProcessedEntity>> {

    @Override
    public List<ProcessedEntity> process(List<YourEntity> items) {
        // 这里写你的批量处理逻辑：比如批量校验、批量转换、批量计算等
        return items.stream()
                .map(item -> {
                    ProcessedEntity processed = new ProcessedEntity();
                    processed.setId(item.getId());
                    processed.setProcessedName(item.getName().toUpperCase());
                    processed.setProcessedValue(item.getValue() * 2);
                    return processed;
                })
                .collect(Collectors.toList());
    }
}

3. 配置Step #

在Step中指定commit-interval为批量大小，确保处理器能接收到对应规模的数据列表：

@Bean
public Step batchProcessingStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                                JdbcCursorItemReader<YourEntity> reader,
                                BatchYourEntityProcessor processor,
                                JdbcBatchItemWriter<ProcessedEntity> writer) {
    return new StepBuilder("batchProcessingStep", jobRepository)
            .<List<YourEntity>, List<ProcessedEntity>>chunk(100, transactionManager)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build();
}

方式二：自定义批量ItemReader直接返回列表 #

如果你希望Reader直接返回完整的数据列表（而非单条数据），可以自定义ItemReader<List<YourEntity>>，每次读取指定数量的数据作为列表返回：

@Bean
public ItemReader<List<YourEntity>> customBatchReader(DataSource dataSource) {
    return new ItemReader<>() {
        private int offset = 0;
        private final int batchSize = 100;

        @Override
        public List<YourEntity> read() throws Exception {
            String sql = "SELECT id, name, value FROM your_table WHERE status = ? LIMIT ? OFFSET ?";
            try (Connection conn = dataSource.getConnection();
                 PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, "PENDING");
                stmt.setInt(2, batchSize);
                stmt.setInt(3, offset);
                ResultSet rs = stmt.executeQuery();
                List<YourEntity> items = new ArrayList<>();
                while (rs.next()) {
                    YourEntity entity = new YourEntity();
                    entity.setId(rs.getLong("id"));
                    entity.setName(rs.getString("name"));
                    entity.setValue(rs.getInt("value"));
                    items.add(entity);
                }
                if (items.isEmpty()) {
                    return null; // 返回null表示读取完毕
                }
                offset += batchSize;
                return items;
            }
        }
    };
}

这种方式下，处理器的实现和方式一完全一致，直接接收列表进行处理。

关键注意事项 #

• PostgreSQL的fetchSize参数需要配合Spring Batch的事务管理生效，无需额外修改数据库的自动提交配置。
• 若处理超大规模数据，优先选择JdbcPagingItemReader分页读取，避免内存溢出。
• 批量处理时要考虑异常处理策略：是整批回滚还是跳过错误数据，可通过SkipPolicy配置实现。

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