Spring Batch 实现千万级表按日分片读写与容错重启的完整实践指南-java教程-PHP中文网

Spring Batch 实现千万级表按日分片读写与容错重启的完整实践指南

本文详解如何利用 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;
    }
}</string></string>

⚠️ 注意：务必使用 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;
}</tablebentity></tablebentity></tableaentity></string></tableaentity></tableaentity>

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，并让框架为您兜底。