本文详解如何用 spring batch 安全、高效地处理 3000 万级记录的跨表同步任务，通过日期维度分片实现并行读取+顺序写入、故障隔离与精准重启，规避内存溢出、状态丢失和无限循环等典型陷阱。
本文详解如何用 spring batch 安全、高效地处理 3000 万级记录的跨表同步任务，通过日期维度分片实现并行读取+顺序写入、故障隔离与精准重启，规避内存溢出、状态丢失和无限循环等典型陷阱。
在企业级数据迁移与 ETL 场景中，面对单表 3000 万行以上的批量处理需求（如从 TableA 按业务日期清洗写入 TableB），简单粗暴的全量扫描极易导致 OOM、事务超时、无法定位失败点、重启成本高等问题。Spring Batch 提供了成熟的分片（Partitioning）机制，但直接套用官方示例常因设计偏差引发“writer 无限循环”“ExecutionContext 膨胀”“chunk 失效”等生产事故——正如提问者所遇困境。本文不讲理论正确性，而是基于真实高负载场景（日均 2800 万订单归档、SLA 可落地、可监控、可运维的工程化方案。
✅ 正确架构：分片 ≠ 并行写入，而是「并行读取 + 顺序提交 + 状态驱动重启」
核心原则是：分区粒度对齐业务语义（按天），但写入必须串行化以保障一致性与可追溯性；失败仅重跑对应日期分区，不波及其他。 这并非 Spring Batch 的“默认行为”，需主动规避两个常见误区：
- ❌ 误区一：在
ItemWriter<list>></list>中直接执行JdbcTemplate.batchUpdate()—— 若未显式控制事务边界，多个分区 Writer 可能并发竞争同一目标表锁，或因异常中断导致部分写入残留； - ❌ 误区二：将
chunk(1)用于 List 批量对象 —— Spring Batch 的 chunk 模型要求ItemReader<t></t>返回单个实体（如TableAEntity），而非List<tableaentity></tableaentity>。强行返回列表会破坏框架对read/process/write生命周期的管理，引发ClassCastException或无限重试。
✅ 正解：采用 JdbcPagingItemReader + 自定义 DateRangePartitioner + CompositeItemWriter 封装顺序写入逻辑，结构如下：
// 1. Reader：按日期范围分页读取（非全加载到内存）
@Bean
public JdbcPagingItemReader<TableAEntity> reader(DataSource dataSource) {
JdbcPagingItemReader<TableAEntity> reader = new JdbcPagingItemReader<>();
reader.setDataSource(dataSource);
reader.setFetchSize(1000); // 控制单次 JDBC 拉取行数
reader.setRowMapper(new BeanPropertyRowMapper<>(TableAEntity.class));
// 关键：使用 date_column BETWEEN ? AND ? 实现安全分片
MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider();
queryProvider.setSelectClause("SELECT id, order_date, amount, status");
queryProvider.setFromClause("FROM table_a");
queryProvider.setWhereClause("WHERE order_date >= :startDate AND order_date < :endDate");
Map<String, Object> parameterValues = new HashMap<>();
parameterValues.put("startDate", LocalDate.now().minusDays(30)); // 由 ExecutionContext 注入
parameterValues.put("endDate", LocalDate.now().minusDays(29));
reader.setParameterValues(parameterValues);
reader.setQueryProvider(queryProvider);
return reader;
}
// 2. Partitioner：生成「日期区间」而非单日（避免小文件效应）
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
Map<String, ExecutionContext> partitions = new HashMap<>();
LocalDate start = this.startDate;
LocalDate end = this.endDate;
long totalDays = ChronoUnit.DAYS.between(start, end);
// 动态计算每片覆盖天数，确保每片约 50–200 万行（根据历史统计调整）
int daysPerPartition = Math.max(1, (int) Math.ceil((double) totalDays / gridSize));
int i = 0;
while (!start.isAfter(end)) {
LocalDate partitionEnd = start.plusDays(daysPerPartition).minusDays(1);
if (partitionEnd.isAfter(end)) partitionEnd = end;
ExecutionContext ctx = new ExecutionContext();
ctx.put("startDate", start.toString()); // 如 "2026-09-01"
ctx.put("endDate", partitionEnd.plusDays(1).toString()); // 开区间，SQL 用 <
partitions.put("partition_" + i++, ctx);
start = partitionEnd.plusDays(1);
}
return partitions;
}✅ 写入层强制顺序化：用 SynchronizedItemWriter 包装批处理
为满足“writer 必须顺序执行”且“不阻塞 reader 并行”的双重约束，禁止在 writer 中开启新事务或调用阻塞操作。推荐方案是：所有分区 writer 共享一个线程安全的队列，由单独的 TaskletStep 消费并批量提交。
// 定义线程安全的写入缓冲区（全局单例）
@Component
public class SequentialBatchWriter {
private final BlockingQueue<List<TableBEntity>> writeQueue = new LinkedBlockingQueue<>();
private final JdbcTemplate jdbcTemplate;
public SequentialBatchWriter(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}
public void submitBatch(List<TableBEntity> batch) {
try {
writeQueue.put(batch); // 非阻塞入队
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Write queue interrupted", e);
}
}
// 由独立 Tasklet 调用（串行执行）
public void flushAll() {
List<List<TableBEntity>> batches = new ArrayList<>();
writeQueue.drainTo(batches);
if (!batches.isEmpty()) {
String sql = "INSERT INTO table_b (order_id, processed_at, amount) VALUES (?, ?, ?)";
List<Object[]> params = batches.stream()
.flatMap(Collection::stream)
.map(e -> new Object[]{e.getOrderId(), e.getProcessedAt(), e.getAmount()})
.collect(Collectors.toList());
jdbcTemplate.batchUpdate(sql, params);
}
}
}
// 在 slaveStep 中注入 writer（注意：不开启事务！）
@Bean
public ItemWriter<TableAEntity> itemWriter(SequentialBatchWriter writer) {
return items -> {
List<TableBEntity> processed = items.stream()
.map(this::customBusinessLogic) // 你的加工逻辑
.collect(Collectors.toList());
writer.submitBatch(processed); // 仅入队，不执行 DB 操作
};
}
// 最后添加一个串行 TaskletStep 消费队列
@Bean
public Step flushStep(SequentialBatchWriter writer) {
return stepBuilderFactory.get("flushStep")
.tasklet((contribution, chunkContext) -> {
writer.flushAll(); // 真正执行 INSERT BATCH
return RepeatStatus.FINISHED;
})
.build();
}⚠️ 关键注意事项与避坑清单
-
Chunk Size 不是“1”或“动态”，而是固定值（建议 1000–5000）：
JdbcPagingItemReader每次read()返回单条记录，chunk size 控制每次process/write批次大小。设为1000可平衡内存占用与网络往返开销，远优于chunk(1)。 -
绝对禁用
@EnableBatchProcessing(modular = true)下的SimpleAsyncTaskExecutor：它会导致 partition handler 创建无界线程池，引发连接耗尽。应显式配置有界线程池：@Bean public TaskExecutorPartitionHandler partitionHandler() { TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler(); handler.setStep(slaveStep()); ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(4); // 根据 CPU 核心数 × 1.5 executor.setMaxPoolSize(8); executor.setQueueCapacity(100); executor.afterPropertiesSet(); handler.setGridSize(8); // 分区数 ≤ 线程池最大数 handler.setTaskExecutor(executor); return handler; } -
重启可靠性依赖
JobRepository持久化：确保JobRepository使用数据库（非内存），否则失败后无法恢复执行上下文。检查BATCH_JOB_EXECUTION_CONTEXT表是否记录各分区的startDate/endDate。 -
监控与诊断：在
StepExecutionListener中记录每个分区处理行数与耗时，写入自定义审计表。当某分区耗时突增 300%，立即触发告警（大概率是该日期数据倾斜或索引失效）。
✅ 总结：这不是“能不能做”，而是“如何稳住”
按日分片不是银弹，但它是在不修改应用代码、不引入 Kafka/Debezium 等中间件前提下，最可控的渐进式大数据同步范式。Spring Batch 的价值不在“自动魔法”，而在其清晰的状态机模型：STARTED → EXECUTING → COMPLETED/FAILED → RESTARTED。只要坚持三条铁律——
① Reader 做范围裁剪（WHERE + 分页），不加载全量；
② Processor 无状态、幂等、轻量（避免远程调用）；
③ Writer 解耦执行与提交，用队列+串行 Tasklet 保序保一致；
——你就能把 3000 万行的“恐怖数字”，拆解为 30 个可预测、可测试、可回滚的原子单元。这正是资深 DBA 与 SRE 共同信奉的工程哲学：复杂系统稳定性的本质，是对不确定性的分而治之。
