本文详解如何利用 spring batch 的分区（partitioning）机制，对 3000 万级数据按日期维度切分、并行读取处理、顺序写入，同时保障失败可精准重启，避免内存溢出与状态失控。
本文详解如何利用 spring batch 的分区（partitioning）机制，对 3000 万级数据按日期维度切分、并行读取处理、顺序写入，同时保障失败可精准重启，避免内存溢出与状态失控。
在处理超大规模数据迁移或 ETL 场景（如从表 A 向表 B 同步约 3000 万条带日期字段的记录）时，简单单步全量处理极易引发 OOM、超时、不可追踪及无法恢复等问题。Spring Batch 原生支持的 分区步（Partitioned Step） 正是为此类场景设计的核心能力——它将逻辑上统一的任务，按业务维度（如 date 列）拆分为多个独立子任务（partitions），每个子任务拥有专属执行上下文、资源隔离和故障边界。
✅ 正确架构：分区 ≠ 并行写入，而是「并行读+串行写」的可控组合
您提出的“按天切分、逐日处理、失败后仅重跑当日”完全符合 Spring Batch 最佳实践。关键在于厘清两个常见误区：
- ❌ 误区一：“Writer 必须全局串行” → 实际上，每个 partition 拥有独立 writer 实例，但可通过共享线程池 + 单线程
JdbcBatchItemWriter或事务协调实现逻辑上的顺序落库； - ❌ 误区二：“Chunk size 必须固定” → 完全可以动态适配：例如设置
chunk(1000)，配合JdbcPagingItemReader按WHERE date = ?分页拉取，单日数据量波动不影响稳定性。
? 核心原则：分区粒度应兼顾并行效率与失败成本。按日切分非常合理——既保证单 partition 数据量可控（通常数万至数十万行），又具备明确业务语义，便于监控、重试与审计。
✅ 推荐实现方案（修正版配置）
1. 优化 Partitioner：生成精确、可复现的分区键
public class DatePartitioner implements Partitioner {
private LocalDate startDate;
private LocalDate endDate;
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
Map<String, ExecutionContext> partitions = new HashMap<>();
LocalDate current = startDate;
int index = 0;
while (!current.isAfter(endDate)) {
ExecutionContext ctx = new ExecutionContext();
ctx.put("targetDate", current.toString()); // 显式字符串化，避免序列化歧义
partitions.put("partition_" + index++, ctx);
current = current.plusDays(1);
}
return partitions;
}
}⚠️ 注意：务必使用
isAfter()而非equals()防止漏掉endDate当日；targetDate存为String更安全（避免LocalDate序列化兼容性问题）。
2. 配置健壮的 Slave Step（关键！）
@Bean
public Step slaveStep() {
return stepBuilderFactory.get("slaveStep")
.<TableAEntity, TableBEntity>chunk(5000) // 合理 chunk size：平衡内存与事务粒度
.reader(reader())
.processor(processor())
.writer(writer())
.faultTolerant()
.skipPolicy(new AlwaysSkipItemSkipPolicy()) // 可选：跳过脏数据
.retry(Exception.class).retryLimit(3) // 可选：网络异常重试
.build();
}
// Reader：基于日期过滤 + 分页，避免全表扫描
@Bean
@StepScope
public JdbcPagingItemReader<TableAEntity> reader(
@Value("#{stepExecutionContext['targetDate']}") String targetDate,
DataSource dataSource) {
MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider();
queryProvider.setSelectClause("SELECT id, name, amount, created_date");
queryProvider.setFromClause("FROM table_a");
queryProvider.setWhereClause("WHERE DATE(created_date) = ?"); // 确保 created_date 有索引
Map<String, Object> parameters = Collections.singletonMap("targetDate", targetDate);
JdbcPagingItemReader<TableAEntity> reader = new JdbcPagingItemReader<>();
reader.setDataSource(dataSource);
reader.setQueryProvider(queryProvider);
reader.setParameterValues(parameters);
reader.setRowMapper(new BeanPropertyRowMapper<>(TableAEntity.class));
reader.setPageSize(1000); // 每次数据库拉取 1000 行
return reader;
}
// Writer：使用 JdbcBatchItemWriter（默认线程安全，自动 batch）
@Bean
@StepScope
public JdbcBatchItemWriter<TableBEntity> writer(DataSource dataSource) {
JdbcBatchItemWriter<TableBEntity> writer = new JdbcBatchItemWriter<>();
writer.setDataSource(dataSource);
writer.setSql("INSERT INTO table_b (id, processed_name, final_amount, batch_date) VALUES (?, ?, ?, ?)");
writer.setItemPreparedStatementSetter((item, ps) -> {
ps.setLong(1, item.getId());
ps.setString(2, item.getProcessedName());
ps.setBigDecimal(3, item.getFinalAmount());
ps.setString(4, item.getBatchDate());
});
return writer;
}3. 主 Job 配置（修复原代码关键缺陷）
@Bean
public Job customerJob() {
return jobBuilderFactory.get("customerJob")
.start(partitioningStep())
.build();
}
@Bean
public Step partitioningStep() {
return stepBuilderFactory.get("partitioningStep")
.partitioner("slaveStep", partitioner()) // 关联自定义 Partitioner
.partitionHandler(partitionHandler()) // 使用 TaskExecutorPartitionHandler
.build();
}
@Bean
public PartitionHandler partitionHandler() {
TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
handler.setStep(slaveStep());
handler.setGridSize(10); // 建议设为 CPU 核心数 * 2，避免过度并发
handler.setTaskExecutor(new SimpleAsyncTaskExecutor("partition-thread-")); // 命名线程便于排查
return handler;
}✅ 关键注意事项与进阶建议
-
重启保障：Spring Batch 自动持久化每个 partition 的
STEP_EXECUTION状态（含READ_COUNT,WRITE_COUNT,COMMIT_COUNT）。若某日失败，重启时仅重跑该 partition，无需人工干预； -
性能调优：
- 确保
table_a.created_date字段有高效索引（如(DATE(created_date))函数索引或直接created_date范围索引）； -
chunk(5000)+pageSize(1000)组合可减少数据库往返，提升吞吐；
- 确保
-
写入顺序控制（如强依赖）：若表 B 严格要求按日期顺序插入（如触发器依赖），可在
writer中添加synchronized块或使用ThreadPoolTaskExecutor配置corePoolSize=1； -
替代方案对比：
-
Tasklet：适合控制流逻辑，但不推荐用于大数据量——您正确指出了stepExecutionContext无法承载百万级对象； -
MultiThreadedStep：虽快但丧失重启能力（所有 chunk 共享同一ExecutionContext），仅适用于“一次成功，否则全量重来”的场景；
-
-
监控与可观测性：启用 Spring Boot Actuator + Batch Metrics，通过
/actuator/batch查看各 partition 执行耗时、失败原因。
✅ 总结
按业务维度（如日期）进行 Spring Batch 分区，是处理亿级数据迁移最成熟、最可控的模式。它天然支持：
✅ 并行加速（读/处理）
✅ 故障隔离（单 partition 失败不影响其他）
✅ 精准重启（自动定位失败 partition）
✅ 资源可控（内存、连接、事务均按 partition 隔离）
放弃“手动管理大数据流”的思路，信任 Spring Batch 的分片调度与状态管理能力——您要做的，是定义清晰的分区策略、编写健壮的 reader/writer，并让框架为您兜底。
