批处理【Spring Batch】
1、Spring Batch 简介
Spring Batch 是一个开源的、全面的、轻量级的批处理框架,通过 Spring Batch 可以实现强大的批处理应用程序的开发。Spring Batch 提供记录/跟踪、 事务管理、作业处理统计、作业重启以及资源管理等功能。Spring Batch 结合定时任务可以发挥更大的作用。
Spring Batch 提供了 ItemReader 、ItemProcessor 和 ItemWriter 来完成数据的读取、处理以及写出操作,并且可以将批处理的执行状态持久化到数据库中。
2、Spring Batch 使用
2.1 创建项目,添加依赖,配置 application.properties
<dependency><groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-batch</artifactId></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-jdbc</artifactId></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId></dependency><dependency> <groupId>com.alibaba</groupId> <artifactId>druid</artifactId> <version>1.1.10</version></dependency><dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId></dependency>
添加数据库相关依赖是为了将批处理的执行状态持久化到数据库中。项目创建完成后,在 application.properties 进行数据库基本信息配置:
spring.datasource.type=com.alibaba.druid.pool.DruidDataSourcespring.datasource.username=rootspring.datasource.password=root123456spring.datasource.url=jdbc:mysql://localhost:3306/test#项目启动时创建数据表的 SQL 脚本,该脚本由 Spring Batch 提供#spring.datasource.schema=classpath:/org/springframework/batch/core/schema-mysql.sql# 在项目启动时执行建表 SQLspring.batch.initialize-schema=always# 禁止 Spring Batch 自动执行,在 SpringBoot 中,默认情况,当项目启动时就会执行配置好的批理操作,添加了该配置后则不会自动执行,而需要用户手动触发执行spring.batch.job.enabled=false
接下来在项目启动类上添加 @EnableBatchProcessing
注解开启 Spring Batch 支持,代码如下:
@SpringBootApplication@EnableBatchProcessingpublic class SpringBatchApplication { public static void main(String[] args) { SpringApplication.run(SpringBatchApplication.class, args); }}
2.2 配置 batch 文件
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.core.configuration.annotation.StepScope;import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;import org.springframework.batch.item.database.JdbcBatchItemWriter;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.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.core.io.ClassPathResource;import javax.sql.DataSource;@Configurationpublic class CsvBatchJobConfig { @Autowired JobBuilderFactory jobBuilderFactory; @Autowired StepBuilderFactory stepBuilderFactory; @Autowired DataSource dataSource; @Bean @StepScope FlatFileItemReader<User> itemReader() { FlatFileItemReader<User> reader = new FlatFileItemReader<>(); reader.setLinesToSkip(1); reader.setResource(new ClassPathResource("data.csv")); reader.setLineMapper(new DefaultLineMapper<User>() {{ setLineTokenizer(new DelimitedLineTokenizer() { { setNames("id", "username", "address", "gender"); //配置列与列之间的间隔符(将通过间隔符对每一行的数据进行切分),最后设置要映射的实体类属性即可 setDelimiter(","); } }); setFieldSetMapper(new BeanWrapperFieldSetMapper<User>() { { setTargetType(User.class); } }); }}); return reader; } @Bean public JdbcBatchItemWriter jdbcBatchItemWriter() { JdbcBatchItemWriter writer = new JdbcBatchItemWriter(); writer.setDataSource(dataSource); writer.setSql("insert into user(id, username,address,gender)" + "values (:id, :username, :address, :gender)"); writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()); return writer; } @Bean Step csvStep() { return stepBuilderFactory.get("csvStep") .<User, User>chunk(2) .reader(itemReader()) .writer(jdbcBatchItemWriter()) .build(); } @Bean Job csvJob() { return jobBuilderFactory.get("csvJob") .start(csvStep()) .build(); }}
创建一个User实体类,并在数据库创建对应表
public class User { private Integer id; private String username ; private String address; private String gender; //省略get/set方法}
2.3 创建 data.csv 文件
id,username,address,gender1,张三,成都,男2,李四,深圳,女3,张三,成都,男4,李四,深圳,女
2.4 创建 controller
@RestControllerpublic class TestController { @Autowired private JobLauncher jobLauncher; @Autowired private Job job; @GetMapping("/hello") public void hello() { try { jobLauncher.run(job, new JobParameters()); } catch (Exception e) { e.printStackTrace(); } }}
JobLauncher 由框架提供, Job 则是刚刚配置的,通过调用 JobLauncher 的 run 方法启动一个批处理。然后启动 Spring Boot 工程并访问 http://localhost:8080/hello
接口,访问成功后, batch 库中会自动创建出多个批处理相关的表,如图 1 所示。这些表用来记录批处理的执行状态,同时 data.csv 中的数据也已经成功插入 user。