Spring Batch(4): Job具體解釋

第四章 配置作業Job

4.1 基本配置

Job的配置有3個必須的屬性。name，jobRepository，steps。一個簡單的Job配置例如以下：

<job id="footballJob">
    <step id="playerload" parent="s1" next="gameLoad"/>
    <step id="gameLoad" parent="s2" next="playerSummarization"/>
    <step id="playerSummarization" parent="s3"/>
</job>

jobRepository默認引用名稱為jobRepository的bean，當然也能夠顯式地配置：

<job id="footballJob" job-repository="specialRepository">
    <step id="playerload" parent="s1" next="gameLoad"/>
    <step id="gameLoad" parent="s3" next="playerSummarization"/>
    <step id="playerSummarization" parent="s3"/>
</job>

4.1.1 Restartable屬性

該屬性定義Job能否夠被重新啟動，默覺得true，在JobExecution運行失敗后，能夠創建還有一個JobExecution來繼續上次的運行。

可是假設該屬性設為false。又一次運行該JobInstance將拋出異常。

<job id="footballJob" restartable="false">
    ...
</job>

4.1.2 攔截Job運行

Spring Batch在Job的生命周期中提供了一些鈎子方法，可這些鈎子方法通過Listener的形式提供。JobListener的接口定義例如以下：

public interface JobExecutionListener {

    void beforeJob(JobExecution jobExecution);

    void afterJob(JobExecution jobExecution);

}

通過實現JobExecutionListener接口並配置給Job，能夠在Job運行前后運行特定的邏輯。

比如在運行結束之后。假設失敗，發送郵件通知管理人員等。

<job id="footballJob">
    <step id="playerload" parent="s1" next="gameLoad"/>
    <step id="gameLoad" parent="s2" next="playerSummarization"/>
    <step id="playerSummarization" parent="s3"/>
    <listeners>
        <listener ref="sampleListener"/>
    </listeners>
</job>

須要注意的是。不管Job是否成功運行，afterJob方法都會運行。Job是否運行成功，能夠從JobExecution中獲取。

public void afterJob(JobExecution jobExecution){
    if( jobExecution.getStatus() == BatchStatus.COMPLETED ){
        //job success
    }
    else if(jobExecution.getStatus() == BatchStatus.FAILED){
        //job failure
    }
}

Listener的運行順序：
beforeJob與配置的順序一樣，afterJob與配置的順序相反。

Listener異常：
Listener的運行過程中假設拋出異常，將導致Job無法繼續完畢，終於狀態為FAILED.因此要合理控制Listener異常對業務的影響。

注解支持：
假設不想使用侵入性強的Listener接口，能夠使用@BeforeJob和@AfterJob兩個注解聲明。

4.1.3 Job抽象與繼承

通用的Job配置能夠抽取出來，作為抽象的Job存在，抽象的Job不同意被實例化：

<job id="baseJob" abstract="true">
    <listeners>
        <listener ref="listenerOne"/>
    <listeners>
</job>

子Job能夠通過繼續共用這些配置（當然。也能夠繼承非抽象的Job）。

<job id="job1" parent="baseJob">
    <step id="step1" parent="standaloneStep"/>

    <listeners merge="true">
        <listener ref="listenerTwo"/>
    <listeners>
</job>

當中的merge=”true”表示合並父job和子job的配置，也就是兩個Listener都生效。同常規的Spring配置。

4.1.4 Job參數驗證

JobParameterValidator組件用於驗證JobParameter。

通過以下配置為job配置驗證器：

<job id="job1" parent="baseJob3">
    <step id="step1" parent="standaloneStep"/>
    <validator ref="paremetersValidator"/>
</job>

4.1.4 屬性的Late Binding

在Spring中。能夠把Bean配置用到的屬性值通過PropertiesPlaceHolderConfiguer把屬性從配置中分離出來獨立管理，理論上來說，在配置Job的時候也能夠使用同樣的方式。可是Spring Batch提供了在運行時配置參數值的能力：

<bean:property name="filePath" value="#{jobParameters['filePath']}" />

在啟動Job時:

    launcher.executeJob("job.xml" , "footjob",
        new JobParametersBuilder().addDate("day", new Date()))
                                  .addString("filePath", "/opt/data/test.xml"));

4.2 配置JobRepository

JobRepository為任務框架中的各個組件對象提供CRUD操作，比如JobExecution，StepExecution。
一個配置樣例例如以下：

<job-repository id="jobRepository" data-source="dataSource" transaction-manager="transactionManager" isolation-level-for-create="SERIALIZABLE" table-prefix="BATCH_" max-varchar-length="1000"/>

4.2.1 事務配置

JobRepository的操作須要事務來保證其完整性以及正確性，這些元數據的完整性對框架來說非常重要。假設沒有事務支持。框架的行為將無法正確定義。
create*方法的事務隔離級別單獨定義，為了保證同一個JobInstance不會被同一時候運行兩次，默認的隔離級別為SERIALIZABLE。能夠被改動：

<job-repository id="jobRepository" isolation-level-for-create="REPEATABLE_READ" />

假設沒有使用Batch命名空間或者沒有使用Factory Bean,則須要顯示配置事務AOP：

<aop:config>
    <aop:advisor  pointcut="execution(* org.springframework.batch.core..*Repository+.*(..))"/>
    <advice-ref="txAdvice" />
</aop:config>

<tx:advice id="txAdvice" transaction-manager="transactionManager">
    <tx:attributes>
        <tx:method name="*" />
    </tx:attributes>
</tx:advice>

4.2.2 表名前綴

默認情況下，Spring Batch須要的表以BATCH作為前綴，只是能夠自己定義：

<job-repository id="jobRepository" table-prefix="e_batch" />

表前綴能夠改動，可是表名和表的列不能被改動。

4.2.3 特殊的Repository

測試環境中。內存級別的數據庫十分方便：

<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
    <property name="transactionManager" ref="transactionManager"/>
</bean>

假設使用的數據庫類型不在SpringBatch的支持中，能夠通過JobRepositoryFactoryBean自己定義。

4.3 配置JobLauncher

默認提供了一個簡單的Launcher:

<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository" />
</bean>

JobLauncher的時序圖例如以下：

假設啟動的請求來自HTTP，那么等待整個Job完畢再返回不是一個好方法。此時須要異步啟動Job，時序圖例如以下：

對應的Launcher配置例如以下：

<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository" />
    <property name="taskExecutor">
        <bean class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
    </property>
</bean>

4.4 運行Job

有多種方式能夠啟動一個Job。可是核心都是通過JobLauncher來實現。
1. 命令行運行
主要通過CommandLineJobRunner類完畢


2. 從Web容器中運行
通過Http請求啟動任務非經常見。時序圖例如以下：

Controller能夠是常規的Spring MVC Controller:

@Controller
public class JobLauncherController {

    @Autowired
    JobLauncher jobLauncher;

    @Autowired
    Job job;

    @RequestMapping("/jobLauncher.html")
    public void handle() throws Exception{
        jobLauncher.run(job, new JobParameters());
    }
}

3. 使用調度框架運行
能夠與其它調度框架一起使用,比如使用Spring的輕量級調用框架Spring Scheduler或者Quartz

4.5 元數據的高級使用方法

除了通過JobRepository對元數據進行CRUD操作外，Spring batch還提供另外的接口用於訪問元數據。
包含： JobExplorer JobOperator。

總體結構例如以下：

4.5.1 JobExplorer

該組件提供了僅僅讀的查詢操作，是JobRepository的僅僅讀版本號，接口定義例如以下：

public interface JobExplorer {

    List<JobInstance> getJobInstances(String jobName, int start, int count);

    JobExecution getJobExecution(Long executionId);

    StepExecution getStepExecution(Long jobExecutionId, Long stepExecutionId);

    JobInstance getJobInstance(Long instanceId);

    List<JobExecution> getJobExecutions(JobInstance jobInstance);

    Set<JobExecution> findRunningJobExecutions(String jobName);
}

配置一個Bean例如以下：

<bean id="jobExplorer" class="org.spr...JobExplorerFactoryBean" p:dataSource-ref="dataSource" />

假設須要制定表名前綴：

<bean id="jobExplorer" class="org.spr...JobExplorerFactoryBean" p:dataSource-ref="dataSource" p:tablePrefix="BATCH_" />

4.5.2 JobOperator

JobOperator集成了非常多接口定義，提供了綜合的操作方法。定義例如以下:

public interface JobOperator {

    List<Long> getExecutions(long instanceId) throws NoSuchJobInstanceException;

    List<Long> getJobInstances(String jobName, int start, int count)
          throws NoSuchJobException;

    Set<Long> getRunningExecutions(String jobName) throws NoSuchJobException;

    String getParameters(long executionId) throws NoSuchJobExecutionException;

    Long start(String jobName, String parameters)
          throws NoSuchJobException, JobInstanceAlreadyExistsException;

    Long restart(long executionId)
          throws JobInstanceAlreadyCompleteException, NoSuchJobExecutionException,
                  NoSuchJobException, JobRestartException;

    Long startNextInstance(String jobName)
          throws NoSuchJobException, JobParametersNotFoundException, JobRestartException,
                 JobExecutionAlreadyRunningException, JobInstanceAlreadyCompleteException;

    boolean stop(long executionId)
          throws NoSuchJobExecutionException, JobExecutionNotRunningException;

    String getSummary(long executionId) throws NoSuchJobExecutionException;

    Map<Long, String> getStepExecutionSummaries(long executionId)
          throws NoSuchJobExecutionException;

    Set<String> getJobNames();

}

配置：

<bean id="jobOperator" class="org.spr...SimpleJobOperator">
    <property name="jobExplorer">
        <bean class="org.spr...JobExplorerFactoryBean">
            <property name="dataSource" ref="dataSource" />
        </bean>
    </property>
    <property name="jobRepository" ref="jobRepository" />
    <property name="jobRegistry" ref="jobRegistry" />
    <property name="jobLauncher" ref="jobLauncher" />
</bean>

當中的startNextInstance方法將使用當前Job的JobParameter。經過JobParametersIncrementer處理之后的參數啟動一個JobInstance。

public interface JobParametersIncrementer {

    JobParameters getNext(JobParameters parameters);

}

以下是一個簡單實現：

public class SampleIncrementer implements JobParametersIncrementer {

    public JobParameters getNext(JobParameters parameters) {
        if (parameters==null || parameters.isEmpty()) {
            return new JobParametersBuilder().addLong("run.id", 1L).toJobParameters();
        }
        long id = parameters.getLong("run.id",1L) + 1;
        return new JobParametersBuilder().addLong("run.id", id).toJobParameters();
    }
}

為job配置incrementer:

<job id="footballJob" incrementer="sampleIncrementer">
    ...
</job>

在每天處理一次的批處理中，Incrementer的實現可能是按日期遞增。