Java：SpringBoot整合Spring Batch示例

文档

• https://docs.spring.io/spring-batch/docs/4.3.9/reference/html/index.html

基础概念

• JobLauncher：作业启动器，启动作业的入口。对应的实现类为SimpleJobLauncher。
• Job：作业，用于配置作业的相关配置，一个作业可以配置多个步骤，步骤之间是有序的。
• Step：步骤，作业具体执行的业务逻辑，一个Job可以配置多个Step。步骤有两种实现方式：
  • Tasklet方式：所有作业逻辑写在一个方法中。
  • Chunk方式：将一个完整的作业逻辑根据作用拆分到三个方法中
    • ItemReader：负责从数据源中读数据（如从文件、数据库等）。
    • ItemProcessor ：负责对读出来的数据进行非法校验和对数据进行加工。
    • ItemWriter：将数据写到某个目标中（如文件、数据库等）。
• JobBuilderFactory：作业构建起工厂，用于构建作业Job对象。
  • get(String name)：设置作业名称。
  • start(Step step)：设置作业启动的第一个步骤。
  • build()：构建Job对象。
• StepBuilderFactory：作业构建器工厂，用于构造步骤Step对象。
  • get(String name)：设置步骤名称。
  • tasklet(Tasklet tasklet)：设置Tasklet。
  • build()：构建Step对象。
• JobRepository：作业持久化，在执行作业的过程中用于操作spring batch相关的表，记录作业的相关状态等。

Tasklet方式示例

运行环境

$ java -version
java version "1.8.0_361"
Java(TM) SE Runtime Environment (build 1.8.0_361-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.361-b09, mixed mode)

spring-batch依赖

<!-- spring-batch -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-batch</artifactId>
</dependency><dependency><groupId>org.springframework.batch</groupId><artifactId>spring-batch-test</artifactId><scope>test</scope>
</dependency><!-- h2、mysql... -->
<dependency><groupId>com.h2database</groupId><artifactId>h2</artifactId><scope>runtime</scope>
</dependency>

备注：第一次照着网上的文章写代码运行，直接报错，原因是缺少引入的依赖；需要引入数据库驱动，如H2或mysql

项目结构

$ tree 
.
├── pom.xml
└── src└── main├── java│   └── com│       └── example│           └── demo│               ├── HelloWorldApplication.java│               ├── config│               │   └── BatchConfig.java│               └── task│                   └── HelloWorldTasklet.java└── resources

完整依赖pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.7</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.example</groupId><artifactId>demo</artifactId><version>0.0.1-SNAPSHOT</version><name>demo</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version><mybatis-plus.version>3.5.2</mybatis-plus.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><!-- spring-batch --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-batch</artifactId></dependency><dependency><groupId>org.springframework.batch</groupId><artifactId>spring-batch-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.h2database</groupId><artifactId>h2</artifactId><scope>runtime</scope></dependency><!-- test --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

编写业务逻辑

package com.example.demo.task;import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.stereotype.Component;/*** 任务*/
@Component
public class HelloWorldTasklet implements Tasklet {@Overridepublic RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {System.out.println("Hello, World!");return RepeatStatus.FINISHED;}
}

配置作业步骤

package com.example.demo.config;import com.example.demo.task.HelloWorldTasklet;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
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;/*** Spring Batch配置*/
@Configuration
public class BatchConfig {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate HelloWorldTasklet helloWorldTasklet;@Beanpublic Step step() {return this.stepBuilderFactory.get("step").tasklet(helloWorldTasklet).build();}@Beanpublic Job job(Step step) {return this.jobBuilderFactory.get("job").start(step).build();}
}

启动类增加注解：@EnableBatchProcessing

package com.example.demo;import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@EnableBatchProcessing
@SpringBootApplication
public class HelloWorldApplication {public static void main(String[] args) {SpringApplication.run(HelloWorldApplication.class, args);}}

Chunk方式示例

项目结构

$ tree
.
├── pom.xml
└── src└── main├── java│   └── com│       └── example│           └── demo│               ├── Application.java│               ├── config│               │   └── BatchConfig.java│               ├── dto│               │   └── Person.java│               └── processor│                   └── PersonItemProcessor.java└── resources└── persons.csv

依赖pom.xml 同上

package com.example.demo.dto;import lombok.Data;@Data
public class Person {private String name;private Integer age;
}

package com.example.demo.config;import com.example.demo.dto.Person;
import com.example.demo.processor.PersonItemProcessor;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
import org.springframework.batch.item.json.JsonFileItemWriter;
import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;/*** Spring Batch配置*/
@Configuration
public class BatchConfig {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Beanpublic Step step() {return this.stepBuilderFactory.get("step").<Person, Person>chunk(10).reader(reader()).processor(processor()).writer(writer()).build();}@Beanpublic Job job(Step step) {return this.jobBuilderFactory.get("job").start(step).build();}/*** 读取csv文件* @return*/@Beanpublic FlatFileItemReader<Person> reader() {return new FlatFileItemReaderBuilder<Person>().name("personItemReader").resource(new ClassPathResource("persons.csv")).delimited().names(new String[] {"name", "age"}).targetType(Person.class).build();}/*** 处理器* @return*/@Beanpublic PersonItemProcessor processor() {return new PersonItemProcessor();}/*** 写到json文件* @return*/@Beanpublic JsonFileItemWriter<Person> writer() {return new JsonFileItemWriterBuilder<Person>().jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>()).resource(new FileSystemResource("persons.json")).name("personItemWriter").build();}}

package com.example.demo.processor;import com.example.demo.dto.Person;
import org.springframework.batch.item.ItemProcessor;public class PersonItemProcessor implements ItemProcessor<Person, Person> {@Overridepublic Person process(Person person) throws Exception {// 为每个对象年龄+1person.setAge(person.getAge() + 1);return person;}
}

启动入口

package com.example.demo;import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@EnableBatchProcessing
@SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}}

启动SpringBoot即可运行任务

读取的persons.csv文件

Tom,18
Jack,20

生成的persons.json 文件

[{"name":"Tom","age":19},{"name":"Jack","age":21}
]

参考文章

1. SpringBatch从入门到实战(二)：HelloWorld
2. Spring Batch 之 Hello World
3. Spring Batch 入门级示例教程
4. Spring Batch 批处理框架优化实践，效率嘎嘎高！