SpringBatch批處理框架：入門項目

1、項目結構如下：

2、文件說明：

2.1、CreditBill：表示信用卡消費記錄領域對象

2.2、CreditBillProcessor：記錄處理類，本場景僅打印信息

2.3、credit-card-bill-201910.csv：原始賬單數據

2.4、job.xml:作業定義文件

2.5、job-context.xml :Spring Batch 批處理任務需要的基礎信息

2.6、JobLaunch：調用批處理作業類

2.7、JobLaunchTest：Junit單元測試類，使用Spring提供的測試框架類。

2.8、pom.xml:引用相關jar包

3、文件內容：

3.1、CreditBill：實體類對象

/**
 * @author miaosj
 * @version 1.0
 * @date 2019/10/8
 */
public class CreditBill {
    /**
     * 銀行賬戶
     */
    private String accountID;
    /**
     * 賬戶名
     */
    private String name;
    /**
     * 消費金額
     */
    private double amount;
    /**
     * 消費日期
     */
    private String date;
    /**
     * 消費場所
     */
    private String address;

    public String getAccountID() {
        return accountID;
    }

    public void setAccountID(String accountID) {
        this.accountID = accountID;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public double getAmount() {
        return amount;
    }

    public void setAmount(double amount) {
        this.amount = amount;
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}

3.2、credit-card-bill-201910.csv：原始賬單數據

4047390012345678,tom,100.00,2013-2-2 12:00:08,Lu Jia Zui road
4047390012345678,tom,320.00,2013-2-3 12:00:08,Lu Jia Zui road
4047390012345678,tom,674.00,2013-2-6 12:00:08,Lu Jia road
4047390012345678,tom,793.00,2013-2-9 12:00:08,Lu Jia Zui road
4047390012345678,tom,360.00,2013-2-11 12:00:08,Lu Jia Zui road
4047390012345678,tom,893.00,2013-2-28 12:00:08,Lu Jia Zui road

3.3、job-context.xml：job基礎設施

<?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-3.0.xsd"
       default-autowire="byName">
    <!--定義作業倉庫SpringBatch提供了兩種作業倉庫來記錄job執行期產生的信息：一種是內存,另一種是數據庫,此處采用內存-->
    <bean id="jobRepository"
          class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
    </bean>
    <!--定義作業調度器,用來啟動Job -->
    <bean id="jobLauncher"
          class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
        <property name="jobRepository" ref="jobRepository"/>
    </bean>
    <!--事務管理器，用於springbatch框架在對數據操作過程中體統事務能力-->
    <bean id="transactionManager"
          class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
</beans>

3.4、job.xml:定義job

<?xml version="1.0" encoding="UTF-8"?>
<bean:beans xmlns="http://www.springframework.org/schema/batch"
            xmlns:bean="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-3.0.xsd
    http://www.springframework.org/schema/batch
    http://www.springframework.org/schema/batch/spring-batch-2.2.xsd">
    <!--引入job-context.xml配置文件-->
    <bean:import resource="classpath:job-context.xml"/>
    <!--定義billJob billStep 包含讀數據  處理數據 寫數據-->
    <job id="billJob">
        <step id="billStep">
            <tasklet transaction-manager="transactionManager">
                <!--commit-interval="2" 表示任務提交間隔的大小 此處表示每處理2條數據 進行一次寫入操作-->
                <chunk reader="csvItemReader" writer="csvItemWriter" processor="creditBillProcessor" commit-interval="2">
                </chunk>
            </tasklet>
        </step>
    </job>

    <!-- 讀取信用卡賬單文件,CSV格式 -->
    <bean:bean id="csvItemReader"
               class="org.springframework.batch.item.file.FlatFileItemReader"
               scope="step">
        <!--設置讀取的文件資源-->
        <bean:property name="resource"
                       value="classpath:data/credit-card-bill-201910.csv"/>
        <!--將文本中的每行記錄轉換為領域對象-->
        <bean:property name="lineMapper">
            <bean:bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
                <!--引用lineTokenizer-->
                <bean:property name="lineTokenizer" ref="lineTokenizer"/>
                <!--fieldSetMapper根據lineTokenizer中定義的names屬性映射到領域對象中去-->
                <bean:property name="fieldSetMapper">
                    <bean:bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
                        <bean:property name="prototypeBeanName" value="creditBill">
                        </bean:property>
                    </bean:bean>
                </bean:property>
            </bean:bean>
        </bean:property>
    </bean:bean>
    <!-- lineTokenizer 定義文本中每行的分隔符號 以及每行映射成FieldSet對象后的name列表 -->
    <bean:bean id="lineTokenizer"
               class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
        <bean:property name="delimiter" value=","/>
        <bean:property name="names">
            <bean:list>
                <bean:value>accountID</bean:value>
                <bean:value>name</bean:value>
                <bean:value>amount</bean:value>
                <bean:value>date</bean:value>
                <bean:value>address</bean:value>
            </bean:list>
        </bean:property>
    </bean:bean>

    <!-- 寫信用卡賬單文件，CSV格式 -->
    <bean:bean id="csvItemWriter"
               class="org.springframework.batch.item.file.FlatFileItemWriter"
               scope="step">
        <!--<bean:property name="resource" value="file:target/data/outputFile.csv"/>-->
        <!--<bean:property name="resource" value="file:target/outputFile.csv"/>-->
        <bean:property name="resource" value="classpath:data/outputFile.csv"/>
        <bean:property name="lineAggregator">
            <bean:bean
                    class="org.springframework.batch.item.file.transform.DelimitedLineAggregator">
                <bean:property name="delimiter" value=","></bean:property>
                <bean:property name="fieldExtractor">
                    <bean:bean
                            class="org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor">
                        <bean:property name="names" value="accountID,name,amount,date,address">
                        </bean:property>
                    </bean:bean>
                </bean:property>
            </bean:bean>
        </bean:property>
    </bean:bean>
    <!--領域對象 並標注為原型-->
    <bean:bean id="creditBill" scope="prototype"
               class="CreditBill">
    </bean:bean>
    <!--負責業務數據的處理-->
    <bean:bean id="creditBillProcessor" scope="step"
               class="CreditBillProcessor">
    </bean:bean>
</bean:beans>

3.5、JobLaunch：java調用

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

/**
 * @author E101206
 * @version 1.0
 * @date 2019/10/8
 */
public class JobLaunch {
    @SuppressWarnings("resource")
    public static void main(String[] args) {
        //初始化應用上下文
        ApplicationContext context = new ClassPathXmlApplicationContext("job/job.xml");
        //獲取作業調度,根據Bean的名稱從Spring的上下文獲取
        JobLauncher launcher = (JobLauncher) context.getBean("jobLauncher");
        //獲取任務對象
        Job job = (Job) context.getBean("billJob");
        try {
            JobExecution result = launcher.run(job, new JobParameters());
            System.out.println(result.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.6、JobLaunchTest：單位測試

/**
 * @author E101206
 * @version 1.0
 * @date 2019/10/8
 */
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"/job/job.xml"})
public class JobLaunchTest {
    @Autowired
    private JobLauncher jobLauncher;

    @Autowired@Qualifier("billJob")
    private Job job;

    @Before
    public void setUp() throws Exception {
    }

    @After
    public void tearDown() throws Exception {
    }

    @Test
    public void billJob() throws Exception {
        JobExecution result = jobLauncher.run(job, new JobParameters());
        System.out.println(result.toString());
    }
}

 3.7、pom.xml

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.msj</groupId>
    <artifactId>spring-batch-example</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <jdk.version>1.8</jdk.version>
        <spring.version>4.3.8.RELEASE</spring.version>
        <spring.batch.version>3.0.7.RELEASE</spring.batch.version>
        <junit.version>4.11</junit.version>
    </properties>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>${jdk.version}</source>
                    <target>${jdk.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <dependencies>
        <!--&lt;!&ndash; Spring Core &ndash;&gt;-->
        <!--<dependency>-->
            <!--<groupId>org.springframework</groupId>-->
            <!--<artifactId>spring-core</artifactId>-->
            <!--<version>${spring.version}</version>-->
        <!--</dependency>-->

        <!--&lt;!&ndash; Spring jdbc, for database &ndash;&gt;-->
        <!--<dependency>-->
            <!--<groupId>org.springframework</groupId>-->
            <!--<artifactId>spring-jdbc</artifactId>-->
            <!--<version>${spring.version}</version>-->
        <!--</dependency>-->

        <!--&lt;!&ndash; Spring XML to/back object &ndash;&gt;-->
        <!--<dependency>-->
            <!--<groupId>org.springframework</groupId>-->
            <!--<artifactId>spring-oxm</artifactId>-->
            <!--<version>${spring.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>


    </dependencies>
</project>

 4、概念

4.1、Job Repository：作業倉庫，負責job、step執行過程中的狀態保存

4.2、Job launcher：作業調度器，提供執行job的入口

4.3、Job：作業，由多個step組成，封裝整個批處理操作

4.4、Step：作業步，job的一個執行環節

4.5、Tasklet：Step中具體執行邏輯的操作，可以重復執行，可以設置具體的同步、異步操作

4.6、Chunk：給定數量的Item的集合，可以定義對Chunk讀操作、處理操作、寫操作，提交間隔等

4.7、Item：一條記錄

4.8、ItemReader：從數據源讀取Item

4.9、ItemProcessor：在Item寫入數據源之前，對數據進行處理如：數據清洗，數據轉換，數據過濾、數據校驗等

4.10、ItemWriter：將Item批量寫入數據源