大家好，又见面了，我是你们的朋友全栈君。

Spring Batch示例教程

欢迎使用Spring Batch示例。Spring Batch是一个用于执行批处理作业的弹簧框架模块。我们可以使用spring批处理来处理一系列作业。

目录[ 隐藏 ]

1 Spring Batch示例

Spring Batch示例

在进行spring批处理示例程序之前，让我们先了解一下spring批处理术语。

工作可以包含’n’个步骤。每个步骤都包含Read-Process-Write任务，或者它可以具有单个操作，称为tasklet。
Read-Process-Write基本上是从数据库，CSV等源读取的，然后处理数据并将其写入数据库，CSV，XML等源。
Tasklet意味着执行单个任务或操作，如清理连接，在处理完成后释放资源。
Read-Process-Write和tasklet可以链接在一起以运行作业。

Spring Batch示例

让我们考虑一个实施弹簧批的实例。我们将考虑以下方案用于实现目的。

包含数据的CSV文件需要转换为XML以及数据和标记将在列名后面命名。

以下是用于弹簧批处理示例的重要工具和库。

Apache Maven 3.5.0 – 用于项目构建和依赖关系管理。
Eclipse Oxygen Release 4.7.0 – 用于创建spring批处理maven应用程序的IDE。
Java 1.8
Spring Core 4.3.12.RELEASE
Spring OXM 4.3.12.RELEASE
Spring JDBC 4.3.12.RELEASE
Spring Batch 3.0.8.RELEASE
MySQL Java Driver 5.1.25 – 基于MySQL的安装使用。这是Spring Batch元数据表所必需的。

Spring Batch示例目录结构

下图说明了Spring Batch示例项目中的所有组件。

弹簧批量示例

Spring Batch Maven依赖项

下面是pom.xml文件的内容，其中包含spring batch示例项目的所有必需依赖项。

 <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 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.journaldev.spring</groupId> <artifactId>SpringBatchExample</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>SpringBatchDemo</name> <url>http://maven.apache.org</url> <properties> <jdk.version>1.8</jdk.version> <spring.version>4.3.12.RELEASE</spring.version> <spring.batch.version>3.0.8.RELEASE</spring.batch.version> <mysql.driver.version>5.1.25</mysql.driver.version> <junit.version>4.11</junit.version> </properties> <dependencies> <!-- Spring Core --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>${spring.version}</version> </dependency> <!-- Spring jdbc, for database --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jdbc</artifactId> <version>${spring.version}</version> </dependency> <!-- Spring XML to/back object --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-oxm</artifactId> <version>${spring.version}</version> </dependency> <!-- MySQL database driver --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>${mysql.driver.version}</version> </dependency> <!-- Spring Batch dependencies --> <dependency> <groupId>org.springframework.batch</groupId> <artifactId>spring-batch-core</artifactId> <version>${spring.batch.version}</version> </dependency> <dependency> <groupId>org.springframework.batch</groupId> <artifactId>spring-batch-infrastructure</artifactId> <version>${spring.batch.version}</version> </dependency> <!-- Spring Batch unit test --> <dependency> <groupId>org.springframework.batch</groupId> <artifactId>spring-batch-test</artifactId> <version>${spring.batch.version}</version> </dependency> <!-- Junit --> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>${junit.version}</version> <scope>test</scope> </dependency> <dependency> <groupId>com.thoughtworks.xstream</groupId> <artifactId>xstream</artifactId> <version>1.4.10</version> </dependency> </dependencies> <build> <finalName>spring-batch</finalName> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-eclipse-plugin</artifactId> <version>2.9</version> <configuration> <downloadSources>true</downloadSources> <downloadJavadocs>false</downloadJavadocs> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> <source>${jdk.version}</source> <target>${jdk.version}</target> </configuration> </plugin> </plugins> </build> </project>

Spring批处理CSV输入文件

以下是用于弹簧批处理的示例CSV文件的内容。

 1001,Tom,Moody, 29/7/2013 1002,John,Parker, 30/7/2013 1003,Henry,Williams, 31/7/2013

Spring批量作业配置

我们必须在配置文件中定义spring bean和spring批处理作业。下面是job-batch-demo.xml文件的内容，它是spring批处理项目中最重要的部分。

 <beans xmlns="http://www.springframework.org/schema/beans" xmlns:batch="http://www.springframework.org/schema/batch" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-3.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd "> <import resource="../config/context.xml" /> <import resource="../config/database.xml" /> <bean id="report" class="com.journaldev.spring.model.Report" scope="prototype" /> <bean id="itemProcessor" class="com.journaldev.spring.CustomItemProcessor" /> <batch:job id="DemoJobXMLWriter"> <batch:step id="step1"> <batch:tasklet> <batch:chunk reader="csvFileItemReader" writer="xmlItemWriter" processor="itemProcessor" commit-interval="10"> </batch:chunk> </batch:tasklet> </batch:step> </batch:job> <bean id="csvFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader"> <property name="resource" value="classpath:csv/input/report.csv" /> <property name="lineMapper"> <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper"> <property name="lineTokenizer"> <bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer"> <property name="names" value="id,firstname,lastname,dob" /> </bean> </property> <property name="fieldSetMapper"> <bean class="com.journaldev.spring.ReportFieldSetMapper" /> <!-- if no data type conversion, use BeanWrapperFieldSetMapper to map by name <bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper"> <property name="prototypeBeanName" value="report" /> </bean> --> </property> </bean> </property> </bean> <bean id="xmlItemWriter" class="org.springframework.batch.item.xml.StaxEventItemWriter"> <property name="resource" value="file:xml/outputs/report.xml" /> <property name="marshaller" ref="reportMarshaller" /> <property name="rootTagName" value="report" /> </bean> <bean id="reportMarshaller" class="org.springframework.oxm.jaxb.Jaxb2Marshaller"> <property name="classesToBeBound"> <list> <value>com.journaldev.spring.model.Report</value> </list> </property> </bean> </beans>

我们正在使用FlatFileItemReader读取CSV文件，CustomItemProcessor处理数据并使用写入XML文件StaxEventItemWriter。
batch:job – 此标记定义我们要创建的作业。Id属性指定作业的ID。我们可以在一个xml文件中定义多个作业。
batch：step – 此标记用于定义spring批处理作业的不同步骤。
Spring Batch Framework提供了两种不同类型的处理方式，即“TaskletStep Oriented”和“Chunk Oriented”。在这个例子中使用的Chunk Oriented风格是指逐个读取数据并在事务边界内创建将被写出的“块”。
读者：用于读取数据的spring bean。我们csvFileItemReader在这个例子中使用了bean作为实例FlatFileItemReader。
处理器：这是用于处理数据的类。我们CustomItemProcessor在这个例子中使用过。
writer：bean用于将数据写入xml文件。
commit-interval：此属性定义处理完成后将提交的块的大小。基本上它意味着ItemReader将逐个读取数据，而ItemProcessor也将以相同的方式处理它，但ItemWriter只有在它等于commit-interval的大小时才会写入数据。
被用作该项目的一部分，是三个重要的接口ItemReader，ItemProcessor并ItemWriter从org.springframework.batch.item包。

Spring批量模型类

首先，我们将CSV文件读入java对象，然后使用JAXB将其写入xml文件。下面是我们的模型类，带有所需的JAXB 注释。

 package com.journaldev.spring.model; import java.util.Date; import javax.xml.bind.annotation.XmlAttribute; import javax.xml.bind.annotation.XmlElement; import javax.xml.bind.annotation.XmlRootElement; @XmlRootElement(name = "record") public class Report { private int id; private String firstName; private String lastName; private Date dob; @XmlAttribute(name = "id") public int getId() { return id; } public void setId(int id) { this.id = id; } @XmlElement(name = "firstname") public String getFirstName() { return firstName; } public void setFirstName(String firstName) { this.firstName = firstName; } @XmlElement(name = "lastname") public String getLastName() { return lastName; } public void setLastName(String lastName) { this.lastName = lastName; } @XmlElement(name = "dob") public Date getDob() { return dob; } public void setDob(Date dob) { this.dob = dob; } @Override public String toString() { return "Report [id=" + id + ", firstname=" + firstName + ", lastName=" + lastName + ", DateOfBirth=" + dob + "]"; } }

请注意，模型类字段应与spring批处理映射器配置中定义的相同，即property name="names" value="id,firstname,lastname,dob"在我们的示例中。

Spring Batch FieldSetMapper

FieldSetMapper转换日期需要自定义。如果不需要数据类型转换，则只BeanWrapperFieldSetMapper应使用自动按名称映射值。

扩展的java类FieldSetMapper是ReportFieldSetMapper。

 package com.journaldev.spring; import java.text.ParseException; import java.text.SimpleDateFormat; import org.springframework.batch.item.file.mapping.FieldSetMapper; import org.springframework.batch.item.file.transform.FieldSet; import org.springframework.validation.BindException; import com.journaldev.spring.model.Report; public class ReportFieldSetMapper implements FieldSetMapper<Report> { private SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy"); public Report mapFieldSet(FieldSet fieldSet) throws BindException { Report report = new Report(); report.setId(fieldSet.readInt(0)); report.setFirstName(fieldSet.readString(1)); report.setLastName(fieldSet.readString(2)); // default format yyyy-MM-dd // fieldSet.readDate(4); String date = fieldSet.readString(3); try { report.setDob(dateFormat.parse(date)); } catch (ParseException e) { e.printStackTrace(); } return report; } }

Spring Batch Item Processor

现在，在作业配置中定义，itemProcessor将在itemWriter之前触发。我们CustomItemProcessor.java为此创建了一个类。

 package com.journaldev.spring; import org.springframework.batch.item.ItemProcessor; import com.journaldev.spring.model.Report; public class CustomItemProcessor implements ItemProcessor<Report, Report> { public Report process(Report item) throws Exception { System.out.println("Processing..." + item); String fname = item.getFirstName(); String lname = item.getLastName(); item.setFirstName(fname.toUpperCase()); item.setLastName(lname.toUpperCase()); return item; } }

我们可以在ItemProcessor实现中操作数据，因为您可以看到我将名字和姓氏值转换为大写。

弹簧配置文件

在我们的spring批量配置文件中，我们已经导入了两个额外的配置文件 – context.xml和database.xml。

 <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-4.3.xsd"> <!-- stored job-meta in memory --> <!-- <bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"> <property name="transactionManager" ref="transactionManager" /> </bean> --> <!-- stored job-meta in database --> <bean id="jobRepository" class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean"> <property name="dataSource" ref="dataSource" /> <property name="transactionManager" ref="transactionManager" /> <property name="databaseType" value="mysql" /> </bean> <bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" /> <bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher"> <property name="jobRepository" ref="jobRepository" /> </bean> </beans>

jobRepository – JobRepository负责将每个Java对象存储到弹簧批处理的正确元数据表中。
transactionManager – 一旦commit-interval的大小和处理的数据相等，它负责提交事务。
jobLauncher – 这是春季批次的核心。此接口包含用于触发作业的run方法。

 <beans xmlns="http://www.springframework.org/schema/beans" xmlns:jdbc="http://www.springframework.org/schema/jdbc" 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-4.3.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.3.xsd"> <!-- connect to database --> <bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource"> <property name="driverClassName" value="com.mysql.jdbc.Driver" /> <property name="url" value="jdbc:mysql://localhost:3306/Test" /> <property name="username" value="test" /> <property name="password" value="test123" /> </bean> <bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" /> <!-- create job-meta tables automatically --> <!-- <jdbc:initialize-database data-source="dataSource"> <jdbc:script location="org/springframework/batch/core/schema-drop-mysql.sql" /> <jdbc:script location="org/springframework/batch/core/schema-mysql.sql" /> </jdbc:initialize-database> --> </beans>

Spring Batch使用一些元数据表来存储批处理作业信息。我们可以从Spring批处理配置中创建它们，但建议通过执行SQL文件手动完成它们，如上面的注释代码所示。从安全的角度来看，最好不要给Spring批处理数据库用户提供DDL执行权限。

Spring批处理表

Spring Batch表与Java中表示它们的Domain对象非常匹配。例如 – JobInstance，JobExecution，JobParameters和StepExecution地图BATCH_JOB_INSTANCE，BATCH_JOB_EXECUTION，分别BATCH_JOB_EXECUTION_PARAMS和BATCH_STEP_EXECUTION。

ExecutionContext映射到BATCH_JOB_EXECUTION_CONTEXT和BATCH_STEP_EXECUTION_CONTEXT。

JobRepository负责将每个java对象保存并存储到正确的表中。

Spring Batch示例教程

以下是每个元数据表的详细信息。

Batch_job_instance：BATCH_JOB_INSTANCE表包含与JobInstance相关的所有信息。
Batch_job_execution_params：BATCH_JOB_EXECUTION_PARAMS表包含与JobParameters对象相关的所有信息。
Batch_job_execution：BATCH_JOB_EXECUTION表保存与JobExecution对象相关的数据。每次运行作业时都会添加一个新行。
Batch_step_execution：BATCH_STEP_EXECUTION表包含与StepExecution对象相关的所有信息。
Batch_job_execution_context：BATCH_JOB_EXECUTION_CONTEXT表保存与Job的ExecutionContext相关的数据。每个JobExecution都只有一个Job ExecutionContext，它包含该特定作业执行所需的所有作业级数据。此数据通常表示在发生故障后必须检索的状态，以便JobInstance可以从失败的位置重新启动。
Batch_step_execution_context：BATCH_STEP_EXECUTION_CONTEXT表保存与Step的ExecutionContext相关的数据。每个StepExecution都只有一个ExecutionContext，它包含了特定步骤执行所需的所有数据。此数据通常表示在发生故障后必须检索的状态，以便JobInstance可以从失败的位置重新启动。
Batch_job_execution_seq：该表保存作业的数据执行顺序。
Batch_step_execution_seq：此表保存步骤执行序列的数据。
Batch_job_seq：此表保存作业序列的数据，以防我们有多个作业，我们将得到多行。

春季批量测试计划

我们的Spring Batch示例项目已准备就绪，最后一步是编写一个测试类来将其作为java程序执行。

 package com.journaldev.spring; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.launch.JobLauncher; import org.springframework.context.support.ClassPathXmlApplicationContext; public class App { public static void main(String[] args) { String[] springConfig = { "spring/batch/jobs/job-batch-demo.xml" }; ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(springConfig); JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher"); Job job = (Job) context.getBean("DemoJobXMLWriter"); JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis()) .toJobParameters(); try { JobExecution execution = jobLauncher.run(job, jobParameters); System.out.println("Exit Status : " + execution.getStatus()); } catch (Exception e) { e.printStackTrace(); } System.out.println("Done"); context.close(); } }

只需运行上面的程序，您将获得如下所示的输出xml。

Copy

<?xml version="1.0" encoding="UTF-8"?><report><record id="1001"><dob>2013-07-29T00:00:00+05:30</dob><firstname>TOM</firstname><lastname>MOODY</lastname></record><record id="1002"><dob>2013-07-30T00:00:00+05:30</dob><firstname>JOHN</firstname><lastname>PARKER</lastname></record><record id="1003"><dob>2013-07-31T00:00:00+05:30</dob><firstname>HENRY</firstname><lastname>WILLIAMS</lastname></record></report>