SpringBatch企業批處理框架Decision流程

以前兩篇介紹瞭如何基本的使用SpringBatch,可是企業應用歷來都不是單一的,一大堆看起來可氣的需求和一堆難以理清的關係.也有可能你的項目尚未Release原始需求已經被改了一大半.而咱們又不肯意更改咱們寫的像藝術品同樣的代碼,最好的狀況也就是在上面加上一些邏輯,而後以熱插拔的方式達到要求.

對於前面介紹的流程也就一個Step,流程圖:

其實SpringBatch是能夠支持多個Step的.如:

對於上面的流程,咱們只須要多定義幾個Step,使用next聯繫起來就能夠工做了.

<job id="exampleJob">
    <step id="StepA" next="StepB"/>
    <step id="StepB" next="StepC"/>
    <step id="StepC"/>
</job>

每一個Step都一個parent屬性,它和Ant的depend有點相似,至關與在執行該Step前執行parent指定的Step.如:

<job id="exampleJob">
    <step id="StepA"/>
    <step id="StepB" parent=」StepA」 next="StepC"/>
    <step id="StepC"/>
</job>

好了,順序的執行永遠是最簡單的,咱們寫程序不可能總順序的,可不能夠來個選擇呢?如:

這個是確定沒有問題的.咱們有兩種方式來進行選擇,一種是使用Decision標記,一種添加Listener.他們均可以憑返回值包含的字符串進行區分來進行選擇流程.

下面我使用Decision作個例子.

Tasklet是Step中被執行的一個單元,Tasklet類:

public class DecisionTasklet  implements Tasklet {

    private String message;

    public DecisionTasklet() {

    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        System.out.println("message is: " + message);
        return RepeatStatus.FINISHED;
    }

    /**
     * @param message
     *            the message to set
     */
    public void setMessage(String message) {
        this.message = message;
    }

Decisor是Step選擇執行的:

public class JobDecider implements JobExecutionDecider {

    public JobDecider() {
    }

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        if(stepExecution.getReadCount() < 1) {
            return FlowExecutionStatus.FAILED;
        }
        return FlowExecutionStatus.COMPLETED;
    }
}

上面的代碼中我 stepExecution.getReadCount() < 1,很明顯,這個示例中我沒有使用任何Reader,所以getReadCount()只會返回0,這裏我只讓他返回FAILED.一下子咱們在註釋掉這段代碼用於測試結果.

Spring Schama:

<?xml version="1.0" encoding="UTF-8"?>
<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/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/batch
	   http://www.springframework.org/schema/batch/spring-batch-2.1.xsd">

    <import resource="applicationBatch.xml"/>

    <bean id="hello" class="net.dbatch.myexample.DecisionTasklet">
        <property name="message" value="hello" />
    </bean>

    <bean id="failed" class="net.dbatch.myexample.DecisionTasklet">
        <property name="message" value="failed" />
    </bean>

    <bean id="success" class="net.dbatch.myexample.DecisionTasklet">
        <property name="message" value="success" />
    </bean>

    <bean id="jobDecider" class="net.dbatch.decision.JobDecider" />

    <batch:job id="decisionJob">
        <batch:step id="step1" next="decision">
            <batch:tasklet ref="hello" transaction-manager="transactionManager" />
        </batch:step>
        <batch:decision id="decision" decider="jobDecider">
            <batch:next on="FAILED" to="step2" />
            <batch:next on="COMPLETED" to="step3" />
        </batch:decision>
        <batch:step id="step2">
            <batch:tasklet ref="failed" transaction-manager="transactionManager" />
        </batch:step>
        <batch:step id="step3">
            <batch:tasklet ref="success" transaction-manager="transactionManager" />
        </batch:step>
    </batch:job>
</beans>

請注意,我把Tasklet類在spring中註冊了3次.

測試類:

public static void main(String[] args) {
        ApplicationContext c = new ClassPathXmlApplicationContext("decision_job.xml");
        SimpleJobLauncher launcher = new SimpleJobLauncher();
        launcher.setJobRepository((JobRepository) c.getBean("jobRepository"));
        launcher.setTaskExecutor(new SyncTaskExecutor());
        try {
            JobExecution je = launcher.run((Job) c.getBean("decisionJob"),
                    new JobParametersBuilder().toJobParameters());
            System.out.println(je);
            System.out.println(je.getJobInstance());
            System.out.println(je.getStepExecutions());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

輸出:

10-22 19:47:13 INFO [job.SimpleStepHandler] - <Executing step: [step1]>
message is: hello
10-22 19:47:13 INFO [job.SimpleStepHandler] - <Executing step: [step2]>
message is: failed

由日誌咱們能夠看出由step1->step2,咱們註釋掉 JobDecider類中的:

if(stepExecution.getReadCount() < 1) {
    return FlowExecutionStatus.FAILED;
}

從新運行測試類.輸出:

10-22 19:53:17 INFO [job.SimpleStepHandler] - <Executing step: [step1]>
message is: hello
10-22 19:53:17 INFO [job.SimpleStepHandler] - <Executing step: [step3]>
message is: success

很明顯它走了step1->step3的流程.

上面講的還有一種執行選擇的流程是添加1個Listener,由於Listener只能使用chunk的方式配置Tasklet,所以示例邏輯上有點羅嗦,我不細述.

假如咱們這樣的Listener:

public class SkipCheckingListener extends StepExecutionListenerSupport {

    public ExitStatus afterStep(StepExecution stepExecution) {
        String exitCode = stepExecution.getExitStatus().getExitCode();
        if (!exitCode.equals(ExitStatus.FAILED.getExitCode()) && 
              stepExecution.getSkipCount() > 0) {
            return new ExitStatus("COMPLETED WITH SKIPS");
        }
        else {
            return null;
        }
    }
}

Spring配置:

<bean id="payStepCheckingListener" class="net.dbatch.sample.PayStepCheckingListener" />

<job id="job">
    <step id="stepA">
        <next on="COMPLETED WITH SKIPS" to="stepB" />
        <next on="FAILED" to="stepC" />
        <next on="*" to="otherStep" />
        <listeners>
            <listener ref="payStepCheckingListener" />
        </listeners>
    </step>
    <step id="stepB" next="stepC" />
    <step id="stepC" />
</job>

next的on屬性能夠匹配簡單的正則表達式.如:

? 能夠匹配0個或一個字符

* 能夠匹配0個或者多個字符

例如:"c*t"將會匹配"cat"和"count", "c?t"只能匹配"cat"可是不能匹配"count".

由Listener的返回值 ExitStatus返回的狀態能夠按照咱們的意願決定下一步執行哪一個Step.