批处理框架 - Spring Batch 简介 - 处理大量数据

Java - Web - Spring 《批处理框架 - Spring Batch 简介 - 处理大量数据》

一、引言

在企业级应用开发中，经常会遇到需要处理大量数据的场景，例如每日数据统计、数据迁移、文件导入导出等。手动编写代码来处理这些大量数据不仅复杂，而且容易出错。Spring Batch 作为 Spring 框架的一个子项目，为开发人员提供了一个轻量级、灵活且强大的批处理解决方案，能够高效地处理大量数据。

二、Spring Batch 概述

Spring Batch 是一个基于 Spring 框架的批处理框架，它提供了大量可重用的组件，如任务调度、数据读取、数据处理、数据写入、事务管理、重试机制等，帮助开发人员快速搭建批处理作业。Spring Batch 可以处理各种数据源，包括数据库、文件、消息队列等。

Spring Batch 的核心概念

• Job（作业）：一个完整的批处理任务，由一个或多个 Step 组成。
• Step（步骤）：Job 的基本执行单元，每个 Step 包含一个 ItemReader、一个 ItemProcessor 和一个 ItemWriter。
• ItemReader（数据读取器）：负责从数据源读取数据。
• ItemProcessor（数据处理器）：对读取的数据进行处理和转换。
• ItemWriter（数据写入器）：将处理后的数据写入目标数据源。

三、Spring Batch 示例：处理大量数据

1. 项目搭建

首先，我们需要创建一个 Spring Boot 项目，并添加 Spring Batch 和相关依赖。在 pom.xml 中添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-batch</artifactId>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

2. 定义数据模型

假设我们要处理用户数据，创建一个 User 类：

public class User {
    private Long id;
    private String name;
    private int age;
    // 构造函数、Getter 和 Setter 方法
    public User() {}
    public User(Long id, String name, int age) {
        this.id = id;
        this.name = name;
        this.age = age;
    }
    public Long getId() {
        return id;
    }
    public void setId(Long id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getAge() {
        return age;
    }
    public void setAge(int age) {
        this.age = age;
    }
    @Override
    public String toString() {
        return "User{" +
                "id=" + id +
                ", name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}

3. 创建 ItemReader

我们使用 FlatFileItemReader 从 CSV 文件中读取用户数据：

import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
@Configuration
public class ReaderConfig {
    @Bean
    public FlatFileItemReader<User> reader() {
        FlatFileItemReader<User> reader = new FlatFileItemReader<>();
        reader.setResource(new ClassPathResource("users.csv"));
        reader.setLineMapper(new DefaultLineMapper<User>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[]{"id", "name", "age"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<User>() {{
                setTargetType(User.class);
            }});
        }});
        return reader;
    }
}

4. 创建 ItemProcessor

创建一个简单的 ItemProcessor，将用户年龄加 1：

import org.springframework.batch.item.ItemProcessor;
import org.springframework.stereotype.Component;
@Component
public class UserProcessor implements ItemProcessor<User, User> {
    @Override
    public User process(User user) throws Exception {
        user.setAge(user.getAge() + 1);
        return user;
    }
}

5. 创建 ItemWriter

使用 ConsoleItemWriter 将处理后的数据输出到控制台：

import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
@Configuration
public class WriterConfig {
    @Bean
    public ItemWriter<User> writer() {
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> items) throws Exception {
                for (User user : items) {
                    System.out.println(user);
                }
            }
        };
    }
}

6. 配置 Job 和 Step

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
@EnableBatchProcessing
public class BatchConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final ItemReader<User> reader;
    private final ItemProcessor<User, User> processor;
    private final ItemWriter<User> writer;
    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                       ItemReader<User> reader, ItemProcessor<User, User> processor, ItemWriter<User> writer) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.reader = reader;
        this.processor = processor;
        this.writer = writer;
    }
    @Bean
    public Step step() {
        return stepBuilderFactory.get("step")
              .<User, User>chunk(10)
              .reader(reader)
              .processor(processor)
              .writer(writer)
              .build();
    }
    @Bean
    public Job job() {
        return jobBuilderFactory.get("job")
              .start(step())
              .build();
    }
}

7. 运行批处理作业

创建一个主类来运行批处理作业：

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class BatchApplication implements CommandLineRunner {
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private Job job;
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }
    @Override
    public void run(String... args) throws Exception {
        JobParameters jobParameters = new JobParametersBuilder()
              .addLong("time", System.currentTimeMillis())
              .toJobParameters();
        jobLauncher.run(job, jobParameters);
    }
}

8. 准备 CSV 文件

在 src/main/resources 目录下创建 users.csv 文件，内容如下：

id,name,age
1,Alice,20
2,Bob,25
3,Charlie,30

9. 运行结果

运行 BatchApplication 类，控制台将输出处理后的用户数据，用户年龄都加了 1：

User{id=1, name='Alice', age=21}
User{id=2, name='Bob', age=26}
User{id=3, name='Charlie', age=31}

四、总结

Spring Batch 是一个强大的批处理框架，通过简单的配置和代码实现，就可以高效地处理大量数据。本文介绍了 Spring Batch 的核心概念，并通过一个简单的示例演示了如何使用 Spring Batch 从 CSV 文件中读取数据、处理数据并将处理后的数据输出到控制台。

核心组件总结

通过学习和使用 Spring Batch，开发人员可以更加轻松地处理企业级应用中的大量数据处理任务。