这里对Spring Batch 进行批处理实践。
介绍
本文将会讲述SpringBatch 如何搭建并运行起来的。
本教程,将会介绍从磁盘读取文件,并写入MySql 中。
什么是Spring Batch
Spring Batch 是Spring的子项目,基于Spring的批处理的框架,通过其可以构建出批量的批处理框架。
官方地址:github.com/spring-projects/spring-batch
入门案例
新建Spring Boot 项目
选择Spring Batch
继续等待
pom 依赖如下
<?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.3.1.RELEASE</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> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-batch</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.batch</groupId> <artifactId>spring-batch-test</artifactId> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build></project>
什么是 Spring Batch works
一个job有读写处理这三个部分组成。
通过JobLauncher启动Job
步骤为 读、处理、写 三个步骤
一个例子
初始化目录结构如下
读取
从数组中读取三个字符
package com.example.demo.step;import org.springframework.batch.item.ItemReader;import org.springframework.batch.item.NonTransientResourceException;import org.springframework.batch.item.ParseException;import org.springframework.batch.item.UnexpectedInputException;public class Reader implements ItemReader<String> { private String[] message = {"ming", "mingming", "mingmingming"}; private int count = 0; @Override public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { if(count < message.length){ return message[count++]; }else{ count = 0; } }}
每次将会调用reader 方法,进行读取一个,
处理
字符串转为大写
package com.example.demo.step;import org.springframework.batch.item.ItemProcessor;public class Processor implements ItemProcessor<String, String> { @Override public String process(String s) throws Exception { return s.toUpperCase(); }}
字符串将会调用其方法将其处理为大写
写
package com.example.demo.step;import org.springframework.batch.item.ItemWriter;import java.util.List;public class Writer implements ItemWriter<String> { @Override public void write(List<? extends String> list) throws Exception { for (String s : list) { System.out.println("Writing the data " + s); } }}
监听
任务成功完成后往控制台输出一行字符串
package com.example.demo.step;import org.springframework.batch.core.BatchStatus;import org.springframework.batch.core.JobExecution;import org.springframework.batch.core.listener.JobExecutionListenerSupport;public class JobCompletionListener extends JobExecutionListenerSupport { @Override public void afterJob(JobExecution jobExecution) { // 项目完成以后调用 if(jobExecution.getStatus() == BatchStatus.COMPLETED){ System.out.println("项目已经完成"); } }}
Config
对项目进行配置
package com.example.demo.config;import com.example.demo.step.JobCompletionListener;import com.example.demo.step.Processor;import com.example.demo.step.Reader;import com.example.demo.step.Writer;import org.springframework.batch.core.Job;import org.springframework.batch.core.JobExecutionListener;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.launch.support.RunIdIncrementer;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class BatchConfig { @Autowired public JobBuilderFactory jobBuilderFactory; @Autowired public StepBuilderFactory stepBuilderFactory; @Bean public Job processJob() { return jobBuilderFactory.get("processJob") .incrementer(new RunIdIncrementer()).listener(listener())// 监听 .flow(orderStep1()).end().build(); // 创建步骤1 } @Bean // 步骤1 bean 先读再写 public Step orderStep1() { return stepBuilderFactory.get("orderStep1").<String, String> chunk(1) .reader(new Reader()).processor(new Processor()) // 读取。处理 .writer(new Writer()).build(); // 最后写 } @Bean public JobExecutionListener listener() { return new JobCompletionListener(); // 创建监听 }}
配置Controller
用来启动应用
package com.example.demo.controller;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.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;@RestControllerpublic class JobInvokerController { @Autowired JobLauncher jobLauncher; @Autowired Job processJob; @RequestMapping("/invokejob") public String handle() throws Exception { JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis()) .toJobParameters(); jobLauncher.run(processJob, jobParameters); return "Batch job has been invoked"; }}
配置文件
spring: batch: job: enabled: false datasource: url: jdbc:h2:file:./DB jpa: properties: hibernate: hbm2ddl: auto: update
Spring Batch在加载的时候job默认都会执行,把spring.batch.job.enabled置为false,即把job设置成不可用,应用便会根据jobLauncher.run来执行。下面2行是数据库的配置,不配置也可以,使用的嵌入式数据库h2
添加注解
Spring Boot入口类:加注解@EnableBatchProcessing
package com.example.demo;import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication@EnableBatchProcessingpublic class DemoApplication { public static void main(String[] args) { SpringApplication.run(DemoApplication.class, args); }}
Run
此时项目run
访问 http://localhost:8080/invokejob 项目已经启动