批处理框架 - 核心组件 - 作业、步骤等组件

Java - Web - Spring 《批处理框架 - 核心组件 - 作业、步骤等组件》

在 Java Web 开发中，Spring Batch 是一个强大的批处理框架，它能够高效地处理大量数据。Spring Batch 提供了一系列核心组件，如作业（Job）、步骤（Step）等，这些组件协同工作，使得批处理任务的开发变得更加简单和高效。本文将详细介绍 Spring Batch 的核心组件，并通过示例代码演示如何使用这些组件。

核心组件概述

作业（Job）

作业是 Spring Batch 中的最高级别抽象，它代表了一个完整的批处理任务。一个作业可以包含一个或多个步骤，并且可以配置作业的启动、停止、重试等行为。作业通常由作业存储库（JobRepository）来管理，作业存储库负责记录作业的执行状态和元数据。

步骤（Step）

步骤是作业的一个独立执行单元，它定义了批处理任务中的一个具体处理步骤。每个步骤可以包含一个或多个任务，如读取数据、处理数据、写入数据等。步骤由步骤执行器（StepExecutor）来执行，并且可以配置步骤的执行顺序、重试策略等。

读取器（ItemReader）

读取器负责从数据源中读取数据，如数据库、文件、消息队列等。Spring Batch 提供了多种内置的读取器，如 JdbcCursorItemReader、FlatFileItemReader 等，同时也支持自定义读取器。

处理器（ItemProcessor）

处理器负责对读取器读取的数据进行处理，如数据转换、数据验证、数据过滤等。处理器可以是一个简单的 Java 类，实现 ItemProcessor 接口。

写入器（ItemWriter）

写入器负责将处理器处理后的数据写入目标数据源，如数据库、文件、消息队列等。Spring Batch 提供了多种内置的写入器，如 JdbcBatchItemWriter、FlatFileItemWriter 等，同时也支持自定义写入器。

演示代码

1. 添加依赖

首先，在 pom.xml 中添加 Spring Batch 和 Spring Boot 的依赖：

<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. 定义实体类

创建一个简单的实体类 Person，用于表示批处理的数据：

public class Person {
    private String firstName;
    private String lastName;
    public Person() {}
    public Person(String firstName, String lastName) {
        this.firstName = firstName;
        this.lastName = lastName;
    }
    // Getters and Setters
    public String getFirstName() {
        return firstName;
    }
    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }
    public String getLastName() {
        return lastName;
    }
    public void setLastName(String lastName) {
        this.lastName = lastName;
    }
    @Override
    public String toString() {
        return "Person [firstName=" + firstName + ", lastName=" + lastName + "]";
    }
}

3. 定义读取器

创建一个 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<Person> reader() {
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        reader.setResource(new ClassPathResource("persons.csv"));
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[] { "firstName", "lastName" });
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});
        }});
        return reader;
    }
}

4. 定义处理器

创建一个 ItemProcessor 来处理读取的数据：

import org.springframework.batch.item.ItemProcessor;
public class PersonItemProcessor implements ItemProcessor<Person, Person> {
    @Override
    public Person process(final Person person) throws Exception {
        final String firstName = person.getFirstName().toUpperCase();
        final String lastName = person.getLastName().toUpperCase();
        return new Person(firstName, lastName);
    }
}

5. 定义写入器

创建一个 JdbcBatchItemWriter 来将处理后的数据写入数据库：

import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;
@Configuration
public class WriterConfig {
    @Bean
    public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        writer.setSql("INSERT INTO persons (first_name, last_name) VALUES (:firstName, :lastName)");
        writer.setDataSource(dataSource);
        return writer;
    }
}

6. 定义步骤和作业

创建一个步骤和作业，并将读取器、处理器和写入器组合在一起：

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.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;
@Configuration
@EnableBatchProcessing
public class BatchConfig {
    @Autowired
    public JobBuilderFactory jobBuilderFactory;
    @Autowired
    public StepBuilderFactory stepBuilderFactory;
    @Autowired
    private ReaderConfig readerConfig;
    @Autowired
    private WriterConfig writerConfig;
    @Bean
    public Step step1(DataSource dataSource) {
        return stepBuilderFactory.get("step1")
              .<Person, Person> chunk(10)
              .reader(readerConfig.reader())
              .processor(new PersonItemProcessor())
              .writer(writerConfig.writer(dataSource))
              .build();
    }
    @Bean
    public Job importUserJob(Step step1) {
        return jobBuilderFactory.get("importUserJob")
              .flow(step1)
              .end()
              .build();
    }
}

7. 运行批处理任务

创建一个 Spring Boot 应用程序，并运行批处理任务：

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.jdbc.core.JdbcTemplate;
import javax.sql.DataSource;
@SpringBootApplication
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }
    @Bean
    public CommandLineRunner run(DataSource dataSource) throws Exception {
        return args -> {
            JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
            jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS persons (id INT AUTO_INCREMENT PRIMARY KEY, first_name VARCHAR(255), last_name VARCHAR(255))");
        };
    }
}

总结

通过以上示例代码，我们演示了如何使用 Spring Batch 的核心组件来创建一个简单的批处理任务。下面是一个表格总结这些核心组件的作用：

Spring Batch 的核心组件提供了强大而灵活的功能，使得批处理任务的开发变得更加简单和高效。希望本文能够帮助你更好地理解和使用 Spring Batch。