springbatch---->springbatch的使用(二)

时间:2023-03-09 08:55:54
springbatch---->springbatch的使用(二)

  这里我们对springbatch做一个比较深入的学习例子,解压文件,读取文件内容过滤写入到数据库中。如果你掉进了黑暗里,你能做的,不过是静心等待,直到你的双眼适应黑暗。

springbatch的使用案例

  首先,我们来列举一下spring batch里面所涉及到的概念。

1、Job repository
An infrastructure component that persists job execution metadata
2、Job launcher
An infrastructure component that starts job executions
3、Job
An application component that represents a batch process
4、Step
A phase in a job; a job is a sequence of steps
5、Tasklet
A transactional, potentially repeatable process occurring in a step
6、Item
A record read from or written to a data source
7、Chunk
A list of items of a given size
8、Item reader
A component responsible for reading items from a data source
9、Item processor
A component responsible for processing (transforming, validating, or filtering) a read item before it’s written
10、Item writer
A component responsible for writing a chunk to a data source

   我们的项目是基于上篇博客的,具体的可以参考博客:springbatch---->springbatch的使用(一)。流程:将得到的zip文件解压成txt文件,根据里面的信息。我们过滤到年龄大于30的数据,并把过滤后的数据按格式存放到数据库中。以下我们只列出新增或者修改的文件内容。

一、关于xml文件的修改,增加一个job.xml文件和修改batch.xml文件

  • batch.xml文件增加内容:
<!-- 读取文件写入到数据库的job -->
<job id="readFlatFileJob">
<step id="decompress" next="readWriter">
<tasklet ref="decompressTasklet"/>
</step>
<step id="readWriter" next="clean">
<tasklet>
<chunk reader="reader" writer="writer" commit-interval="100" processor="processor"/>
</tasklet>
</step>
<step id="clean">
<tasklet ref="cleanTasklet"/>
</step>
</job>

  关于commit-interval:Number of items to process before issuing a commit. When the number of items read reaches the commit interval number, the entire corresponding chunk is written out through the item writer and the transaction is committed.

  • 新增的job.xml文件内容如下:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <!-- 对压缩文件进行解压 -->
<bean id="decompressTasklet" class="spring.batch.readFile.DecompressTasklet">
<property name="inputResource" value="file:file/file.zip"/>
<property name="targetDirectory" value="file"/>
<property name="targetFile" value="file.txt"/>
</bean> <!-- 读取文本文件 -->
<bean id="reader" class="org.springframework.batch.item.file.FlatFileItemReader">
<property name="lineMapper" ref="lineMapper"/>
<property name="linesToSkip" value="1"/> <!-- 跳过第一行,也就是从第二行开始读 -->
<property name="resource" value="file:file/file.txt"/>
</bean> <!-- 对读取的内容做过滤处理 -->
<bean id="processor" class="spring.batch.readFile.FileProcessor"/> <!-- 将读取的内容写到数据库中 -->
<bean id="writer" class="spring.batch.readFile.FileWriter">
<constructor-arg ref="dataSource"/>
</bean> <!-- 清除压缩文件-->
<bean id="cleanTasklet" class="spring.batch.readFile.FileCleanTasklet">
<property name="resource" value="file:file/file.zip"/>
</bean> <!-- 对文件里面的数据进行|分割,映射成一个类 -->
<bean id="lineMapper" class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<property name="lineTokenizer" ref="delimitedLineTokenizer"/>
<property name="fieldSetMapper">
<bean class="spring.batch.readFile.ReadFileMapper"/>
</property>
</bean>
<bean id="delimitedLineTokenizer" class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<constructor-arg name="delimiter" value="|"/>
</bean> <!-- 简单的数据源配置 -->
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://127.0.0.1:3306/csiilearn"/>
<property name="username" value="root"/>
<property name="password" value="*****"/>
</bean>
</beans>

其实关于有格式的文件的读写,springbatch为我们提供了FlatFileItemWriter和FlatFileItemReader类。我们可以直接使用,只需要配置它的属性就行。

二、新增加的java类,如下所示

springbatch---->springbatch的使用(二)

  • DecompressTasklet:对zip文件进行解压
package spring.batch.readFile;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
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.core.io.Resource; import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.util.zip.ZipInputStream; /**
* @Author: huhx
* @Date: 2017-11-01 上午 9:39
*/
public class DecompressTasklet implements Tasklet {
private Resource inputResource;
private String targetDirectory;
private String targetFile; @Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
System.out.println("file path: " + inputResource.getFile().getAbsolutePath());
ZipInputStream zis = new ZipInputStream(
new BufferedInputStream(
inputResource.getInputStream()));
File targetDirectoryAsFile = new File(targetDirectory);
if (!targetDirectoryAsFile.exists()) {
FileUtils.forceMkdir(targetDirectoryAsFile);
}
File target = new File(targetDirectory, targetFile);
BufferedOutputStream dest = null;
while (zis.getNextEntry() != null) {
if (!target.exists()) {
target.createNewFile();
}
FileOutputStream fos = new FileOutputStream(target);
dest = new BufferedOutputStream(fos);
IOUtils.copy(zis, dest);
dest.flush();
dest.close();
}
zis.close();
if (!target.exists()) {
throw new IllegalStateException("Could not decompress anything from the archive!");
}
return RepeatStatus.FINISHED;
} public void setInputResource(Resource inputResource) {
this.inputResource = inputResource;
} public void setTargetDirectory(String targetDirectory) {
this.targetDirectory = targetDirectory;
} public void setTargetFile(String targetFile) {
this.targetFile = targetFile;
}
}
  • People:数据库映射的javaBean类
public class People implements Serializable {
private String username;
private int age;
private String address;
private Date birthday; ..get...set...
}
  • ReadFileMapper:读取文件字段的映射
package spring.batch.readFile;

import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.validation.BindException; /**
* @Author: huhx
* @Date: 2017-11-01 上午 10:11
*/
public class ReadFileMapper implements FieldSetMapper<People> { @Override
public People mapFieldSet(FieldSet fieldSet) throws BindException {
People people = new People();
people.setUsername(fieldSet.readString(0));
people.setAge(fieldSet.readInt(1));
people.setAddress(fieldSet.readString(2));
people.setBirthday(fieldSet.readDate(3));
return people;
}
}
  • FileProcessor:对读取的数据做进一步的处理,这里我们是过滤操作
package spring.batch.readFile;

import org.springframework.batch.item.ItemProcessor;

/**
* @Author: huhx
* @Date: 2017-11-01 上午 9:43
*/
public class FileProcessor implements ItemProcessor<People, People> {
@Override
public People process(People item) throws Exception {
return needsToBeFiltered(item) ? null : item;
} private boolean needsToBeFiltered(People item) {
int age = item.getAge();
return age > 30 ? true : false;
}
}
  • FileWriter:对最终过滤后的数据写入数据库中
package spring.batch.readFile;

import org.springframework.batch.item.ItemWriter;
import org.springframework.jdbc.core.JdbcTemplate; import javax.sql.DataSource;
import java.util.List; /**
* @Author: huhx
* @Date: 2017-11-01 上午 9:42
*/
public class FileWriter implements ItemWriter<People> {
private static final String INSERT_PRODUCT = "INSERT INTO batch_user (user_name, age, address, birthday) VALUES(?, ?, ?, ?)"; private JdbcTemplate jdbcTemplate; public FileWriter(DataSource dataSource) {
this.jdbcTemplate = new JdbcTemplate(dataSource);
} @Override
public void write(List<? extends People> items) throws Exception {
for (People people : items) {
jdbcTemplate.update(INSERT_PRODUCT, people.getUsername(), people.getAge(), people.getAddress(), people.getBirthday());
}
}
}
  • FileCleanTasklet:做流程最后的清理处理,删除zip文件
package spring.batch.readFile;

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.core.io.Resource; import java.io.File; /**
* @Author: huhx
* @Date: 2017-11-01 上午 9:44
*/
public class FileCleanTasklet implements Tasklet {
private Resource resource; public void setResource(Resource resource) {
this.resource = resource;
} @Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
File file = resource.getFile();
file.deleteOnExit();
return RepeatStatus.FINISHED;
}
}

三、修改JobLaunch.java文件里面的内容

Job job = (Job) context.getBean("readFlatFileJob");
  • zip解压后的文件内容:
姓名|年龄|地址|生日
李元芳||黄冈|--
王昭君||武汉|--
狄仁杰||天津|--
孙悟空||黄冈|--
牛魔王||武汉|--
孙尚香||天津|--
  • 运行后的数据库batch_user表的数据:

springbatch---->springbatch的使用(二)

友情链接