Spring Batch

Spring Batch 是 Spring 生态体系中的一个轻量级、全面的批处理框架，专门用于高效处理大规模数据集（如百万/亿级记录）的批量任务。它提供了可扩展的架构、事务管理、错误恢复和任务调度等核心功能，广泛应用于金融、物流、电商等需要定期执行离线数据处理的场景。以下从核心概念、架构、功能、典型场景及代码示例展开说明：

一、核心概念

1. 批处理（Batch Processing）
   批处理是一种非交互式的数据处理方式，通过一次性处理大量数据（如文件、数据库记录）生成结果或报表，而非实时响应单条请求。例如：

   • 银行每日生成交易对账单
   • 电商每日统计销售数据
   • 定期清理历史数据

2. Spring Batch 的定位

   • 轻量级：基于 Spring Framework，无需额外依赖复杂中间件。
   • 企业级：支持事务、重试、并行处理、监控等生产级特性。
   • 可扩展：通过组件化设计（如 ItemReader、ItemProcessor、ItemWriter）适配不同数据源和业务逻辑。

二、核心架构

Spring Batch 采用分层架构，主要组件包括：

1. JobRepository

   • 存储批处理任务的元数据（如任务状态、执行记录），支持内存（MapJobRepository）或数据库（JDBC）持久化。

2. JobLauncher

   • 启动批处理任务（Job），支持参数传递。

3. Job

   • 封装整个批处理流程，由多个 Step 组成。

4. Step

   • 批处理的最小执行单元，包含：
     • ItemReader：读取数据（如从文件、数据库）。
     • ItemProcessor：处理数据（如转换、校验）。
     • ItemWriter：写入数据（如保存到数据库、生成文件）。
   • 支持顺序执行或条件跳转（通过 Flow）。

5. Chunk（分块处理）

   • 核心优化机制：批量读取数据（如每次1000条），处理后批量写入，减少数据库I/O。

三、核心功能

1. 事务管理

   • 自动管理 Step 内的事务边界，确保数据一致性（如处理1000条记录时，若第500条失败，可回滚整个Chunk）。

2. 错误处理与重试

   • 通过 SkipPolicy 跳过可忽略错误（如空值），或通过 RetryPolicy 重试失败记录。

3. 并行处理

   • 支持多线程（TaskletStep）或分区（PartitionStep）处理大数据集。

4. 任务调度

   • 结合 Spring Scheduler 或 Quartz 实现定时任务（如每日凌晨执行）。

5. 监控与日志

   • 提供 JobExplorer 查询任务历史，或通过 Spring Boot Actuator 暴露监控指标。

四、典型应用场景

1. 数据迁移

   • 将旧系统数据批量导入新系统（如 MySQL → HBase）。

2. 报表生成

   • 每日统计用户行为数据，生成Excel/PDF报表。

3. 数据清洗

   • 批量处理用户上传的CSV文件，过滤无效数据并写入数据库。

4. 定时任务

   • 每月自动计算员工工资，生成工资单。

五、代码示例

以下是一个简单的 Spring Batch 任务示例：从 CSV 文件读取数据，处理后写入数据库。

1. 定义实体类

public class User {
    private String id;
    private String name;
    // getters/setters...
}

2. 配置批处理任务

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private DataSource dataSource;

    // 1. 定义 ItemReader：从CSV读取数据
    @Bean
    public FlatFileItemReader<User> userItemReader() {
        return new FlatFileItemReaderBuilder<User>()
            .name("userItemReader")
            .resource(new ClassPathResource("users.csv"))
            .delimited()
            .names(new String[]{"id", "name"})
            .targetType(User.class)
            .build();
    }

    // 2. 定义 ItemProcessor：处理数据（示例：转换名称格式）
    @Bean
    public ItemProcessor<User, User> userItemProcessor() {
        return user -> {
            user.setName(user.getName().toUpperCase());
            return user;
        };
    }

    // 3. 定义 ItemWriter：写入数据库
    @Bean
    public JdbcBatchItemWriter<User> userItemWriter() {
        return new JdbcBatchItemWriterBuilder<User>()
            .dataSource(dataSource)
            .sql("INSERT INTO users (id, name) VALUES (:id, :name)")
            .beanMapped()
            .build();
    }

    // 4. 定义 Step：分块处理（Chunk=10）
    @Bean
    public Step userImportStep() {
        return stepBuilderFactory.get("userImportStep")
            .<User, User>chunk(10)
            .reader(userItemReader())
            .processor(userItemProcessor())
            .writer(userItemWriter())
            .build();
    }

    // 5. 定义 Job：包含一个Step
    @Bean
    public Job importUserJob() {
        return jobBuilderFactory.get("importUserJob")
            .start(userImportStep())
            .build();
    }
}

6. 启动任务

@SpringBootApplication
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
        // 或通过 JobLauncher 动态启动
    }
}

六、总结

• 优势：Spring Batch 提供了完整的批处理解决方案，简化开发复杂度，尤其适合需要高吞吐量、高可靠性的数据批量处理场景。
• 适用性：若任务是定时、非实时且数据量较大（如每日百万级），Spring Batch 是首选；若需实时处理，应考虑消息队列（如 Kafka）或流处理框架（如 Flink）。
• 扩展：可结合 Spring Cloud Data Flow 实现分布式批处理，或使用 Spring Batch Admin 进行可视化监控。

通过合理设计 Step 和 Chunk，Spring Batch 能高效处理从简单到复杂的批处理需求。