Spring Batch：大数据批量处理实战

1. 核心概念

Spring Batch 是轻量级批处理框架，专为处理大规模数据作业设计，核心组件包括：

• 作业(Job)：完整批处理任务，由多个步骤组成
• 步骤(Step)：作业的独立执行单元
• 读-处理-写模式：
  • ItemReader：数据输入（如数据库、文件）
  • ItemProcessor：业务逻辑处理
  • ItemWriter：结果输出

2. 架构优势

graph LR
A[启动器JobLauncher] --> B[作业Job]
B --> C[步骤Step1]
B --> D[步骤Step2]
C --> E[Reader->Processor->Writer]
D --> F[Reader->Processor->Writer]

• 事务管理：支持块级事务处理
• 重启能力：故障后从断点继续执行
• 扩展性：通过分区(Partitioning)实现并行处理

3. 实战示例：CSV文件处理

// 配置作业
@Bean
public Job csvProcessingJob(JobBuilderFactory jobs, Step step1) {
    return jobs.get("csvJob")
        .incrementer(new RunIdIncrementer())
        .flow(step1)
        .end()
        .build();
}

// 配置步骤
@Bean
public Step step1(StepBuilderFactory steps, 
                 ItemReader<Person> reader,
                 ItemProcessor<Person, Person> processor,
                 ItemWriter<Person> writer) {
    
    return steps.get("step1")
        .<Person, Person>chunk(100)  // 每100条数据提交事务
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .build();
}

// 自定义处理器
public class PersonProcessor implements ItemProcessor<Person, Person> {
    @Override
    public Person process(Person item) {
        item.setName(item.getName().toUpperCase());  // 姓名转大写
        return item;
    }
}

4. 关键配置项

5. 性能优化策略

1. 分区处理：将数据拆分为子集并行执行

   @Bean
   public Partitioner partitioner() {
       return new MultiResourcePartitioner();  // 多文件分区
   }
   

2. 异步执行：使用AsyncItemProcessor+AsyncItemWriter
3. 批处理参数：

   spring.batch.jdbc.initialize-schema=always  # 自动初始化表
   spring.batch.job.enabled=false              # 禁止自动启动作业
   

6. 错误处理机制

.skipLimit(10)              // 允许跳过10条错误数据
.skip(DataFormatException.class) 
.retryLimit(3)             // 重试3次
.retry(DeadlockLoserDataAccessException.class)

7. 典型应用场景

• 金融领域：日终对账处理
• 电商系统：每日库存同步
• 日志分析：TB级日志清洗
• 数据迁移：旧系统到新系统的数据转移

最佳实践：处理1亿条数据时，通过分区+异步处理可将原需8小时的任务缩短至35分钟（需根据硬件配置调整块大小和线程数）

通过合理配置事务块大小、线程池参数及错误处理策略，Spring Batch 可稳定处理TB级数据任务，是企业级批处理的首选框架。