深入解析 Spring Batch：架构、性能优化与高级应用

hanweiweihanweiwei
5 min read

1. Spring Batch 简介

Spring Batch 是一个专为大规模数据处理设计的框架，提供了事务管理、并发执行、日志追踪、重试/跳过、分片等功能，确保批处理任务的高效和可靠。通过 Spring Batch，开发者可以轻松实现复杂的批处理任务，处理海量数据，同时保证数据的一致性和系统的可扩展性。

大数据处理的一般处理流程：

graph LR
    K[数据读取]--> M[数据处理]
    M --> N[处理结果存储]

2. Spring Batch 核心架构

graph LR
    A[应用程序层] --> B[核心层]
    B --> C[基础设施层]

Spring Batch 的核心架构围绕 Job 和 Step 进行设计，其组件包括：

  • Job：一个完整的批处理任务。

  • Step：Job 的最小执行单元。

  • ItemReader：从数据源读取数据。

  • ItemProcessor：处理数据，例如转换、过滤。

  • ItemWriter：写入目标数据源。

  • JobRepository：存储任务的元数据。

  • JobLauncher：触发任务执行。

@Bean
public Job userExportJob(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, DataSource dataSource) {
    // Step 1: 读取数据
    ItemReader<User> userReader = new JdbcCursorItemReader<>();
    ((JdbcCursorItemReader<User>) userReader).setDataSource(dataSource);
    ((JdbcCursorItemReader<User>) userReader).setSql("SELECT id, name FROM users");

    // Step 2: 处理数据
    ItemProcessor<User, User> userProcessor = user -> {
        user.setName(user.getName().toUpperCase()); // 转换名字为大写
        return user;
    };

    // Step 3: 写入数据
    ItemWriter<User> userWriter = new FlatFileItemWriter<>();
    ((FlatFileItemWriter<User>) userWriter).setResource(new FileSystemResource("users.csv"));

    // Step 定义
    Step readProcessWriteStep = stepBuilderFactory.get("readProcessWriteStep")
            .<User, User>chunk(10)
            .reader(userReader)
            .processor(userProcessor)
            .writer(userWriter)
            .build();

    // 创建 Job
    return jobBuilderFactory.get("userExportJob")
            .start(readProcessWriteStep)
            .build();
}

解释：

  • ItemReader：JdbcCursorItemReader 用于从数据库中逐条读取用户数据。

  • ItemProcessor：将每个用户的名字转换为大写。

  • ItemWriter：FlatFileItemWriter 将处理后的用户数据写入 CSV 文件。

  • Chunk：每 10 条数据作为一个 chunk 进行处理。每处理一个 chunk，系统会将数据从内存写入磁盘，释放内存，减少内存使用和提高处理效率。

流程：

  1. 读取：ItemReader 从数据库中读取一批数据（这里是每 10 条）。

  2. 处理：ItemProcessor 对每条数据进行处理（例如，将用户名转为大写）。

  3. 写入：ItemWriter 将处理后的数据写入目标文件或数据库。

通过这种分块的方式，Spring Batch 能够高效地处理大规模数据，同时确保每个块的数据在处理完后能够被写入和清理，从而避免内存溢出。

3.基本组件

classDiagram
    direction LR
    class Job {
        +String name
        +List<JobInstance> jobInstances
        +void launch()
    }

    class JobInstance {
        +String id
        +Date createTime
        +List<JobExecution> jobExecutions
        +JobExecution getJobExecution()
    }

    class JobExecution {
        +String status
        +Date startTime
        +Date endTime
        +JobExecutionContext context
        +void execute()
    }

    class JobExecutionContext {
        +Map<String, Object> data
        +void put(String key, Object value)
        +Object get(String key)
    }

    Job "1" --> "0..*" JobInstance : contains
    JobInstance "1" --> "0..*" JobExecution : contains
    JobExecution "1" --> "1" JobExecutionContext : has

  • Job：

    • 表示一个批处理作业，拥有多个 JobInstance。

    • 包含 launch() 方法来启动任务。

  • JobInstance：

    • 表示一次批处理作业的实例，每个 JobInstance 可能有多个 JobExecution。

    • 包含 getJobExecution() 方法来获取该实例的执行记录。

  • JobExecution：

    • 记录作业执行的状态、开始和结束时间等信息。

    • 执行时，JobExecution 会创建并关联一个 JobExecutionContext。

  • JobExecutionContext：

    • 存储作业执行过程中的详细信息（如数据处理进度或错误信息），通过 put() 和 get() 方法进行数据存取。
graph LR
    A[Job: userExportJob] 
    A1[Job Instance: userExportJob-2023]
    A2[Job Instance: userExportJob-2024]

    A1_1[Job Execution: Started]
    A1_2[Job Execution: Completed]
    A2_1[Job Execution: Started]
    A2_2[Job Execution: Failed]

    A1_1_1[Job Execution Context: Data Processed]
    A1_2_1[Job Execution Context: Data Processed]
    A2_1_1[Job Execution Context: Data Processed]
    A2_2_1[Job Execution Context: Error Details]

    A --> A1
    A --> A2

    A1 --> A1_1
    A1 --> A1_2
    A2 --> A2_1
    A2 --> A2_2

    A1_1 --> A1_1_1
    A1_2 --> A1_2_1
    A2_1 --> A2_1_1
    A2_2 --> A2_2_1

4. Spring Batch 的执行流程

Spring Batch 采用 Chunk-Oriented Processing（块处理）模式，每个 Step 以 Chunk 为单位读取、处理并写入数据。

  1. JobLauncher 启动 Job。

  2. JobRepository 记录任务状态。

  3. Step 逐块读取数据，每次读取 chunkSize 条。

  4. ItemProcessor 处理每个数据项。

  5. ItemWriter 批量提交数据，减少数据库交互。

  6. JobRepository 记录执行进度，支持失败重试。

graph TD
    A[开始] --> B[ItemReader：读取数据（从数据库）]
    B --> C[Chunk：分块处理（每 100 条数据为一个 Chunk）]
    C --> D[ItemProcessor：处理数据（例如：转换格式）]
    D --> E[ItemWriter：写入数据（例如：CSV文件）]
    E --> F[事务提交（每个 Chunk 提交）]
    F --> G{是否还有更多数据?}
    G -->|是| B
    G -->|否| H[结束]

    subgraph Transaction [事务处理]
        B
        D
        E
    end

    classDef startend fill:#81C784,stroke:#388E3C;
    classDef process fill:#FFEB3B,stroke:#FF9800;
    classDef readwrite fill:#81D4FA,stroke:#03A9F4;
    classDef decision fill:#FFCDD2,stroke:#F44336;
    classDef transaction fill:#FFEBEE,stroke:#FF9800,stroke-width:2px;

    class A,H startend;
    class B,C,D,E process;
    class F readwrite;
    class G decision;
    class Transaction transaction;

5.流程控制

5.1. Flow（流程）：

在 Spring Batch 中，Flow 是指一系列的 Step 或 Tasklet，它们会根据一定的顺序执行。流控制允许更复杂的任务执行逻辑。

graph TD
    A[开始] --> B[Step 1]
    B --> C[Step 2]
    C --> D[Step 3]
    D --> E[结束]

    classDef startend fill:#81C784,stroke:#388E3C;
    classDef process fill:#FFEB3B,stroke:#FF9800;

    class A,E startend;
    class B,C,D process;

说明：

  • Flow 允许多个 Step 顺序执行，从 Step 1 开始到 Step 3 结束。

  • 这适用于没有条件判断的简单顺序任务。

5.2. Step（步骤）：

在 Spring Batch 中，Step 是处理单个任务的基本单位，它可以包括读取、处理和写入的操作。

graph TD
    A[开始] --> B[ItemReader：读取数据]
    B --> C[ItemProcessor：处理数据]
    C --> D[ItemWriter：写入数据]
    D --> E[结束]

    classDef startend fill:#81C784,stroke:#388E3C;
    classDef process fill:#FFEB3B,stroke:#FF9800;

    class A,E startend;
    class B,C,D process;

说明：

  • Step 包含了从读取数据到处理数据再到写入数据的完整流程，执行这些任务的顺序不能更改。

  • 每个 Step 都可以独立运行，也可以在 Flow 中与其他 Step 组合。

5.3. Partition（分区）：

分区处理是为了并行化批处理任务，将一个大的数据集分成多个小块进行处理。每个块被分配给不同的线程或节点进行并行执行。

graph TD
    A[开始] --> B[Partition：分区读取]
    B --> C[Step 1：处理分区 1]
    B --> D[Step 2：处理分区 2]
    B --> E[Step 3：处理分区 3]
    C --> F[合并结果]
    D --> F
    E --> F
    F --> G[结束]

    classDef startend fill:#81C784,stroke:#388E3C;
    classDef process fill:#FFEB3B,stroke:#FF9800;
    classDef partition fill:#FFCCBC,stroke:#FF5722;

    class A,G startend;
    class B partition;
    class C,D,E,F process;

说明：

  • 在 Partition 中，数据被分成几个部分，每个分区可以并行处理。

  • 每个分区对应一个 Step，处理完成后，结果会被合并。

5.4. Decision（决策）：

Decision 用于在批处理过程中，根据条件判断决定接下来执行哪个 Step。例如，基于某个条件判断是否继续执行某个任务。

graph TD
    A[开始] --> B[Step 1：读取数据]
    B --> C[Decision：条件判断]
    C -->|条件A成立| D[Step 2：执行任务A]
    C -->|条件B成立| E[Step 3：执行任务B]
    D --> F[结束]
    E --> F

    classDef startend fill:#81C784,stroke:#388E3C;
    classDef process fill:#FFEB3B,stroke:#FF9800;
    classDef decision fill:#FFCDD2,stroke:#F44336;

    class A,F startend;
    class B,C process;

说明：

  • Decision 步骤根据某个条件来决定接下来的流程：

    • 如果条件 A 满足，则执行 Step 2；

    • 如果条件 B 满足，则执行 Step 3。

  • 这种决策控制适用于处理需要根据上下文或数据动态选择执行路径的任务。

总结：

  • Flow 控制任务按顺序执行。

  • Step 是最基本的执行单元，完成一系列处理任务。

  • Partition 用于并行化数据处理，将数据分割成多个块进行并行处理。

  • Decision 用于根据条件动态地决定下一步执行的 Step。

6. Spring Batch 事务管理与错误处理

6.1 事务回滚与隔离级别

Spring Batch 允许对 Step 级别或 Chunk 级别设置事务：

@Bean
public Step transactionalStep() {
    return stepBuilderFactory.get("transactionalStep")
            .<String, String>chunk(10)
            .reader(reader())
            .processor(processor())
            .writer(writer())
            .transactionManager(transactionManager())
            .faultTolerant()
            .retryLimit(3)
            .skip(Exception.class)
            .skipLimit(5)
            .build();
}

  • retryLimit(3)：失败重试 3 次。

  • skipLimit(5)：跳过最多 5 个错误项。

6.2 断点续传与失败恢复

JobRepository 记录任务状态，允许失败后从上次中断点继续执行：

@Bean
public Job job(Step step) {
    return jobBuilderFactory.get("restartableJob")
            .incrementer(new RunIdIncrementer())
            .start(step)
            .build();
}

7. 并发优化与性能提升

7.1 多线程并发处理

利用 TaskExecutor 并行执行 Step：

@Bean
public TaskExecutor taskExecutor() {
    SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("batch-task");
    executor.setConcurrencyLimit(5);
    return executor;
}

@Bean
public Step multiThreadedStep() {
    return stepBuilderFactory.get("multiThreadedStep")
            .<String, String>chunk(10)
            .reader(reader())
            .processor(processor())
            .writer(writer())
            .taskExecutor(taskExecutor())
            .build();
}

7.2 Partitioning 分区处理

适用于大规模数据处理，通过 Partitioner 分片并行执行 Step。

@Bean
public Step partitionedStep() {
    return stepBuilderFactory.get("partitionedStep")
            .partitioner("slaveStep", partitioner())
            .step(slaveStep())
            .gridSize(4)
            .taskExecutor(taskExecutor())
            .build();
}

7.3 数据库批量写入优化

避免单条 SQL 执行，使用批量提交优化性能。

@Bean
public ItemWriter<String> writer(DataSource dataSource) {
    JdbcBatchItemWriter<String> writer = new JdbcBatchItemWriter<>();
    writer.setDataSource(dataSource);
    writer.setSql("INSERT INTO batch_table (data) VALUES (:data)");
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
    return writer;
}

8. JSON 大数据处理优化

对于 JSON 解析，使用 JacksonStreaming 处理大文件，避免内存溢出。

@Bean
public JsonItemReader<MyObject> jsonReader() {
    return new JsonItemReaderBuilder<MyObject>()
            .jsonObjectReader(new JacksonJsonObjectReader<>(MyObject.class))
            .resource(new FileSystemResource("data.json"))
            .name("jsonReader")
            .build();
}

9. 总结

Spring Batch 通过 Job 和 Step 进行任务拆分，并支持事务管理、并发执行、批量提交和流式处理，适用于大规模数据处理场景。合理利用多线程、分区处理和流式 JSON 解析，可大幅提升性能

0
Subscribe to my newsletter

Read articles from hanweiwei directly inside your inbox. Subscribe to the newsletter, and don't miss out.

Springbootspringbatch

Written by

hanweiwei
hanweiwei