Spring batch是用来处理大量数据操作的一个框架，主要用来读取大量数据，然后进行一定处理后输出成指定的形式。
Spring batch主要有以下部分组成：
- JobRepository 用来注册job的容器
- JobLauncher 用来启动Job的接口
- Job 实际执行的任务，包含一个或多个Step
- Step step包含ItemReader、ItemProcessor和ItemWriter
- ItemReader 用来读取数据的接口
- ItemProcessor 用来处理数据的接口
- ItemWriter 用来输出数据的接口
以上Spring Batch的主要组成部分只需要注册成Spring的Bean即可。若想开启批处理的支持还需在配置类上使用@EnableBatchProcessing，在Spring Batch中提供了大量的ItemReader和ItemWriter的实现，用来读取不同的数据来源，数据的处理和校验都要通过ItemProcessor接口实现来完成。
Spring Boot的支持
Spring Boot对Spring Batch支持的源码位于org.springframework.boot.autoconfigure.batch下。
Spring Boot为我们自动初始化了Spring Batch存储批处理记录的数据库。
spring batch会自动加载hsqldb驱动，根据需求选择去留。
下面是一个spring boot支持spring batch 的例子：
1. 实体类
public class Person {
@Size(max=4,min=2) //使用JSR-303注解来校验注解
private String name;
private int age;
private String nation;
private String address;
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;
}
public String getNation() {
return nation;
}
public void setNation(String nation) {
this.nation = nation;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
}2. 校验器
public class CsvBeanValidator<T> implements Validator<T>,InitializingBean {
private javax.validation.Validator validator;
@Override
public void afterPropertiesSet() throws Exception { //使用JSR-303的Validator来校验我们的数据，在此处进行JSR-303的Validator的初始化
ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
validator = validatorFactory.usingContext().getValidator();
}
@Override
public void validate(T value) throws ValidationException {
Set<ConstraintViolation<T>> constraintViolations = validator.validate(value); //使用Validator的validate方法校验数据
if(constraintViolations.size()>0){
StringBuilder message = new StringBuilder();
for (ConstraintViolation<T> constraintViolation : constraintViolations) {
message.append(constraintViolation.getMessage() + "\n");
}
throw new ValidationException(message.toString());
}
}
}3. ItemProcessor
public class CsvItemProcessor extends ValidatingItemProcessor<Person>{
@Override
public Person process(Person item) throws ValidationException {
super.process(item); //需要执行super.process(item)才会调用自定义校验器
if(item.getNation().equals("汉族")){ //对数据做简单的处理，若民族为汉族，则数据转换成01，其余转换成02
item.setNation("01");
}else{
item.setNation("02");
}
return item;
}
}4. Job监听（监听器要实现JobExecutionListener接口，并重写其beforeJob、afterJob方法即可）
public class CsvJobListener implements JobExecutionListener{
long startTime;
long endTime;
@Override
public void beforeJob(JobExecution jobExecution) {
startTime = System.currentTimeMillis();
System.out.println("任务处理开始");
}
@Override
public void afterJob(JobExecution jobExecution) {
endTime = System.currentTimeMillis();
System.out.println("任务处理结束");
System.out.println("耗时:" + (endTime - startTime) + "ms");
}
}5. 配置
@Configuration
@EnableBatchProcessing
public class CsvBatchConfig { @Bean
public ItemReader<Person> reader() throws Exception {
FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>(); //使用FlatFileItemReader读取文件
reader.setResource(new ClassPathResource("people.csv")); //使用FlatFileItemReader的setResource方法设置CSV文件的路径
reader.setLineMapper(new DefaultLineMapper<Person>() {{ //在此处对CVS文件的数据和领域模型类做对应映射
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames(new String[] { "name","age", "nation" ,"address"});
}});
setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
setTargetType(Person.class);
}});
}});
return reader;
} @Bean
public ItemProcessor<Person, Person> processor() {
CsvItemProcessor processor = new CsvItemProcessor(); //使用自定义的ItemProcessor的实现
processor.setValidator(csvBeanValidator()); //为Processor指定校验器
return processor;
} @Bean
public ItemWriter<Person> writer(DataSource dataSource) {//Spring能让容器中已有的Bean以参数的形式注入，Spring boot已经定义了DataSource
JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<Person>(); //使用JDBC批处理的JdbcBatchItemWriter来写数据到数据库
writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
String sql = "insert into person " + "(id,name,age,nation,address) "
+ "values(hibernate_sequence.nextval, :name, :age, :nation,:address)";
writer.setSql(sql); //在此设置要执行批处理的sql语句
writer.setDataSource(dataSource);
return writer;
} @Bean
public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager)
throws Exception {
JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
jobRepositoryFactoryBean.setDataSource(dataSource);
jobRepositoryFactoryBean.setTransactionManager(transactionManager);
jobRepositoryFactoryBean.setDatabaseType("oracle");
return jobRepositoryFactoryBean.getObject();
} @Bean
public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager)
throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
return jobLauncher;
} @Bean
public Job importJob(JobBuilderFactory jobs, Step s1) {
return jobs.get("importJob")
.incrementer(new RunIdIncrementer())
.flow(s1) //指定step
.end()
.listener(csvJobListener()) //绑定监听器
.build();
} @Bean
public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader, ItemWriter<Person> writer,
ItemProcessor<Person,Person> processor) {
return stepBuilderFactory
.get("step1")
.<Person, Person>chunk(65000) //批处理每次提交65000条数据
.reader(reader) //给step绑定reader
.processor(processor) //给step绑定Processor
.writer(writer) //给step绑定writer
.build();
} @Bean
public CsvJobListener csvJobListener() {
return new CsvJobListener();
} @Bean
public Validator<Person> csvBeanValidator() {
return new CsvBeanValidator<Person>();
} }
6.application.xml
spring.datasource.driverClassName=oracle.jdbc.OracleDriver
spring.datasource.url=jdbc\:oracle\:thin\:@localhost\:1521\:xe
spring.datasource.username=boot
spring.datasource.password=boot spring.batch.job.enabled=true logging.level.org.springframework.web = DEBUG
上面的例子是自动触发批处理的，当我们需要手动触发批处理时，需要将CsvBatchConfig类的@Configuration注解注释掉，让此配置类不再起效，新建TriggerBatchConfig配置类，内容与CsvBatchConfig完全一致，除了修改定义ItemReader这个Bean;另外，还需要修改application.xml配置文件spring.batch.job.enable=false
@Configuration
@EnableBatchProcessing
public class TriggerBatchConfig { @Bean
@StepScope
public FlatFileItemReader<Person> reader(@Value("#{jobParameters['input.file.name']}") String pathToFile) throws Exception {
FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>(); //
reader.setResource(new ClassPathResource(pathToFile)); //
reader.setLineMapper(new DefaultLineMapper<Person>() {{ //
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames(new String[] { "name","age", "nation" ,"address"});
}});
setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
setTargetType(Person.class);
}});
}}); return reader;
} @Bean
public ItemProcessor<Person, Person> processor() {
CsvItemProcessor processor = new CsvItemProcessor();
processor.setValidator(csvBeanValidator());
return processor;
} @Bean
public ItemWriter<Person> writer(DataSource dataSource) {
JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<Person>();
writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
String sql = "insert into person " + "(id,name,age,nation,address) "
+ "values(hibernate_sequence.nextval, :name, :age, :nation,:address)";
writer.setSql(sql); //
writer.setDataSource(dataSource);
return writer;
} @Bean
public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager)
throws Exception {
JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
jobRepositoryFactoryBean.setDataSource(dataSource);
jobRepositoryFactoryBean.setTransactionManager(transactionManager);
jobRepositoryFactoryBean.setDatabaseType("oracle");
return jobRepositoryFactoryBean.getObject();
} @Bean
public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager)
throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
return jobLauncher;
} @Bean
public Job importJob(JobBuilderFactory jobs, Step s1) {
return jobs.get("importJob")
.incrementer(new RunIdIncrementer())
.flow(s1)
.end()
.listener(csvJobListener())
.build();
} @Bean
public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader, ItemWriter<Person> writer,
ItemProcessor<Person,Person> processor) {
return stepBuilderFactory
.get("step1")
.<Person, Person>chunk(65000)
.reader(reader)
.processor(processor)
.writer(writer)
.build();
} @Bean
public CsvJobListener csvJobListener() {
return new CsvJobListener();
} @Bean
public Validator<Person> csvBeanValidator() {
return new CsvBeanValidator<Person>();
} }
控制层代码
@RestController
public class DemoController { @Autowired
JobLauncher jobLauncher; @Autowired
Job importJob;
public JobParameters jobParameters; @RequestMapping("/read")
public String imp(String fileName) throws Exception{ String path = fileName+".csv";
jobParameters = new JobParametersBuilder()
.addLong("time", System.currentTimeMillis())
.addString("input.file.name", path)
.toJobParameters();
jobLauncher.run(importJob,jobParameters);
return "ok";
} }