spring batch教程之配置并运行Job

た入场券 2022-06-18 00:25 449阅读 0赞

Spring batch整体的架构设计使用如下关系图来进行表示：

![20170328164027418][]

虽然Job对象看上去像是对于多个Step的一个简单容器，但是开发者必须要注意许多配置项。此外，Job的运行以及Job运行过程中元数据如何被保存也是需要考虑的。本章将会介绍Job在运行时所需要注意的各种配置项。

# **1.1 Configuring a Job** #

Job接口 的实现有多个，但是在配置上命名空间存在着不同。必须依赖的只有三项：名称 name，JobRespository 和 Step的列表：

在这个例子中使用了父类的bean定义来创建step，更多描述step配置的信息可以参考step configuration这一节。XML命名空间默认会使用id为'jobRepository'的引用来作为repository的定义。然而可以向如下显式的覆盖：

此外，job配置的step还包含其他的元素，有并发处理()，显示的流程控制()和外化的流程定义()。

## **1.1.1 Restartablity** ##

执行批处理任务的一个关键问题是要考虑job被重启后的行为。如果一个 JobExecution 已经存在一个特定的 JobInstance，那么这个job启动时可以认为是“重启”。 理想情况下，所有任务都能够在他们中止的地方启动，但是有许多场景这是不可能的。在这种场景中就要有开发者来决定创建一个新的 JobInstance ，Spring对此也提供了一些帮助。如果job不需要重启，而是总是作为新的 JobInstance 来运行，那么可重启属性可以设置为'false'：

设置重启属性restartable为‘false’表示‘这个job不支持再次启动’，重启一个不可重启的job会抛出JobRestartExceptio的异常：

Job job = new SimpleJob();
       job.setRestartable(false);
       JobParameters jobParameters = new JobParameters();
       JobExecution firstExecution = jobRepository.createJobExecution(job, jobParameters);
       jobRepository.saveOrUpdate(firstExecution);
       try {
           jobRepository.createJobExecution(job, jobParameters);
           fail();
        } catch (JobRestartException e) {
    //预计抛出JobRestartException异常
        }

这个JUnit代码展示了创建一个不可重启的Job后，第一次能够创建 JobExecution ，第二次再创建相同的JobExcution会抛出一个 JobRestartException。

## **1.1.2 Intercepting Job Execution** ##

在job执行过程中，自定义代码能够在生命周期中通过事件通知执行会是很有用的。SimpleJob能够在适当的时机调用JobListener：

public interface JobExecutionListener {
      void beforeJob(JobExecution jobExecution);
      void afterJob(JobExecution jobExecution);
    }

JobListener能够添加到SimpleJob中去，作为job的listener元素：

无论job执行成功或是失败都会调用afterJob，都可以从 JobExecution 中获取运行结果后，根据结果来进行不同的处理：

public void afterJob(JobExecution jobExecution){
        if( jobExecution.getStatus() == BatchStatus.COMPLETED ){
           //job执行成功 }
        else if(jobExecution.getStatus() == BatchStatus.FAILED){
           //job执行失败 }
    }

对应于这个interface的annotation为：

*  @BeforeJob
 *  @AfterJob

## **1.1.3 Inheriting from a parent Job** ##

如果一组job配置共有相似，但又不是完全相同，那么可以定义一个"父”job，让这些job去继承属性。同Java的类继承一样，子job会把父job的属性和元素合并进来。  
下面的例子中，“baseJob”是一个抽象的job定义，只定义了一个监听器列表。名为“job1”的job是一个具体定义，它继承了“baseJob"的监听器，并且与自己的监听器合并，最终生成的job带有两个监听器，以及一个名为”step1“的step。

## **1.1.4 JobParametersValidator** ##

一个在xml命名空间描述的job或是使用任何抽象job子类的job，可以选择为运行时为job参数定义一个验证器。在job启动时需要保证所有必填参数都存在的场景下，这个功能是很有用的。有一个DefaultJobParametersValidator可以用来限制一些简单的必选和可选参数组合，你也可以实现接口用来处理更复杂的限制。验证器的配置支持使用xml命名空间来作为job的子元素，例如：

验证器可以作为一个引用(如上)来定义也可以直接内嵌定义在bean的命名空间中。

#    **1.2 Java Config** #

在Spring 3版本中可以采用java程序来配置应用程序，来替代XML配置的方式。 正如在Spring Batch 2.2.0版本中，批处理任务中可以使用相同的java配置项来对其进行配置。关于Java的基础配置的两个组成部分分别是： @EnableBatchConfiguration注释和两个builder。

在Spring的体系中 @EnableBatchProcessing 注释的工作原理与其它的带有 @Enable \* 的注释类似。在这种情况  
下， @EnableBatchProcessing 提供了构建批处理任务的基本配置。在这个基本的配置中，除了创建了一个 StepScope 的实例，还可以将一系列可用的bean进行自动装配:

*  JobRepository bean 名称 "jobRepository"
 *  JobLauncher bean名称"jobLauncher"
 *  JobRegistry bean名称"jobRegistry"
 *  PlatformTransactionManager bean名称 "transactionManager"
 *  JobBuilderFactory bean名称"jobBuilders"
 *  StepBuilderFactory bean名称"stepBuilders"

这种配置的核心接口是 BatchConfigurer。它为以上所述的bean提供了默认的实现方式，并要求在context中提供一个bean，即 DataSource 。数据库连接池由被 JobRepository 使用。

注意 只有一个配置类需要有@ enablebatchprocessing注释。只要有一个类添加了这个注释，则以上所有的bean都是可以使用的。

在基本配置中，用户可以使用所提供的builder factory来配置一个job。下面的例子是通过 JobBuilderFactory 和  
StepBuilderFactory 配置的两个step job 。

@Configuration
    @EnableBatchProcessing
    @Import(DataSourceCnfiguration.class)
    public class AppConfig {
       @Autowired
       private JobBuilderFactory jobs;
       @Autowired
       private StepBuilderFactory steps;
       @Bean
       public Job job() {
           return jobs.get("myJob").start(step1()).next(step2()).build();
       }
       @Bean
       protected Step step1(ItemReader<Person> reader, ItemProcessor<Person, Person> processor, ItemWriter<Person> writer) {
             return steps.get("step1")
               .<Person, Person> chunk(10)
               .reader(reader)
               .processor(processor)
               .writer(writer)
               .build();
          }
       @Bean
       protected Step step2(Tasklet tasklet) {
            return steps.get("step2")
               .tasklet(tasklet)
               .build();
          }
    }

# **1.3 Configuring a JobRepository** #

之前说过，JobRepository 是基本的CRUD操作，用于持久化Spring Batch的领域对象(如JobExecution,StepExecution)。许多主要的框架组件(如JobLauncher,Job,Step)都需要使用JobRepository。batch的命名空间中已经抽象走许多JobRepository的实现细节，但是仍然需要一些配置：

<job-repository id="jobRepository"

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

上面列出的配置除了id外都是可选的。如果没有进行参数配置，默认值就是上面展示的内容，之所以写出来是用于展示给读者。 max-varchar-length 的默认值是2500，这表示varchar列的长度，在 sample schema scripts 中用于存储类似于 exit code 这些描述的字符。如果你不修改schema并且也不会使用多字节编码，那么就不用修改它。

## **1.3.1 JobRepository 的事物配置** ##

如果使用了namespace，repository会被自动加上事务控制，这是为了确保批处理操作元数据以及失败后重启的状态能够被准确的持久化，如果repository的方法不是事务控制的，那么框架的行为就不能够被准确的定义。 create\* 方法的隔离级别会被单独指定，为了确保任务启动时，如果两个操作尝试在同时启动相同的任务，那么只有一个任务能够被成功启动。这种方法默认的隔离级别是 SERIALIZABLE ，这是相当激进的做法： READ\_COMMITED 能达到同样效果；如果两个操作不以这种方式冲突的话 READ\_UNCOMMITED 也能很好工作。但是，由于调用 create\* 方法是相当短暂的，只要数据库支持，就不会对性能产生太大影响。它也能被这样覆盖：

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

如果factory的namespace没有被使用，那么可以使用AOP来配置repository的事务行为：

<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>

这个配置片段基本上可以不做修改直接使用。记住加上适当的namespace描述去确保spring-tx和spring-aop(或是整个spring)都在classpath中。

##    **1.3.2 修改 Table 前缀** ##

JobRepository 可以修改的另一个属性是元数据表的表前缀。默认是以BATCH\_开头， BATCH\_JOB\_EXECUTION 和  
BATCH\_STEP\_EXECUTION 就是两个例子。但是，有一些潜在的原因可能需要修改这个前缀。例如schema的名字需要被预置到表名中，或是不止一组的元数据表需要放在同一个schema中，那么表前缀就需要改变：

<job-repository id="jobRepository" table-prefix="SYSTEM.TEST_" />

按照上面的修改配置，每一个元数据查询都会带上 SYSTEM.TEST\_ 的前缀， BATCH\_JOB\_EXECUTION 将会被更换为 SYSTEM.TEST\_JOB\_EXECUTION 。

注意：表名前缀是可配置的，表名和列名是不可配置的。

## **1.3.3 In-Memory Repository** ##

有的时候不想把你的领域对象持久化到数据库中，可能是为了运行的更快速，因为每次提交都要开销额外的时间；也可能并不需要为特定任务保存状态。那么Spring Batch还提供了内存Map版本的job仓库：

需要注意的是 内存 Repository 是轻量的并且不能在两个JVM实例间重启任务，也不能允许同时启动带有相同参数的任务，不适合在多线程的任务或是一个本地分片任务的场景下使用。而使用数据库版本的Repository则能够拥有这些特性。

但是也需要定义一个事务管理器，因为仓库需要回滚语义，也因为商业逻辑要求事务性（例如RDBMS访问）。经过测试许多人觉得 ResourcelessTransactionManager 是很有用的。

##    **1.3.4 Non-standard Database Types in a Repository** ##

如果使用的数据库平台不在支持的平台列表中，在SQL类型类似的情况下你可以使用近似的数据库类型。使用原生的JobRepositoryFactoryBean 来取代命名空间缩写后设置一个相似的数据库类型：

(如果没有指定 databaseType ，JobRepositoryFactoryBean 会通过DataSource自动检测数据库的类型).平台之间的主要不同之处在于主键的计算策略，也可能需要覆盖 incrementerFactory (使用Spring Framework提供的标准实现)。 如果它还不能工作，或是你不使用RDBMS，那么唯一的选择是让 SimpleJobRepository 使用Spring方式依赖并且绑定在手工实现的各种Dao接口上。

#    **1.4 Configuring a JobLauncher** #

JobLauncher 最基本的实现是 SimpleJobLauncher ，它唯一的依赖是通过 JobRepository 获取一个 execution：

一旦获取到 JobExecution ，那么可以通过执行 Job 的方法，最终将 JobExecution 返回给调用者.

从调度启动时，整个序列能够很好的直接工作，但是，从HTTP请求中启动则会出现一些问题。在这种场景中，启动任务需要异步操作，让SimpleJobLauncher能够立刻返回结果给调用者，如果让HTTP请求一直等待很长时间知道批处理任务完成获取到执行结果，是很糟糕的操作体验。一个流程如下图所示：

![20170328164128889][]

通过配置 TaskExecutor 可以很容易的将 SimpleJobLauncher 配置成异步操作：

TaskExecutor 接口的任何实现都能够用来控制 job 的异步执行。

#    **1.5 Running a Job** #

运行一个批处理任务至少有两点要求：一个 JobLauncher 和一个用来运行的 job 。它们都包含了相同或是不同的 context 。举例来说，从命令行来启动job，会为每一个job初始化一个JVM，因此每个job会有一个自己的 JobLauncher；从web容器的HttpRequest来启动job，一般只是用一个 JobLauncher 来异步启动job，http请求会调用这个 JobLauncher 来启动它们需要的job。

##    **1.5.1 在 Web Container 内部运行 Jobs** ##

过去，像批处理任务这样的离线计算都需要从命令行启动。但是，许多例子(包括报表、点对点任务和web支持)都表明，从HttpRequest启动是一个更好的选择。另外，批处理任务一般都是需要长时间运行，异步启动时最为重要的：

![20170328164154202][]

这个例子中的Controller就是spring MVC中的Controller(Spring MVC的信息可以在[http://docs.spring.io/spring/docs/3.2.x/spring-framework-reference/html/mvc.html][http_docs.spring.io_spring_docs_3.2.x_spring-framework-reference_html_mvc.html] 中查看)。Controller通过使用配置为异步的(asynchronously)JobLauncher启动job后立即返回了JobExecution。job保持运行，这个非阻塞的行为能够让controller在持有HttpRequest时立刻返回。示例如下：

@Controller
    public class JobLauncherController {
        @Autowired
        JobLauncher jobLauncher;
        @Autowired
        Job job;
        @RequestMapping("/jobLauncher.html")
        public void handle() throws Exception{
          jobLauncher.run(job, new JobParameters());
        }
    }

# **1.6 Meta-Data 高级用法** #

到目前为止，已经讨论了 JobLauncher 和 JobRepository 接口，它们展示了简单启动任务，以及批处理领域对象的基本CRUD操作：

![20170328164215703][]

一个JobLauncher使用一个JobRepository创建并运行新的JobExection对象，Job和Step实现随后使用相同的JobRepository在job运行期间去更新相同的JobExecution对象。这些基本的操作能够满足简单场景的需要，但是对于有着数百个任务和复杂定时流程的大型批处理情况来说，就需要使用更高级的方式访问元数据：

![20170328164233297][]

接下去会讨论 JobExplorer 和 JobOperator 两个接口，能够使用更多的功能去查询和修改元数据。

## **1.6.1 Querying the Repository** ##

在使用高级功能之前，需要最基本的方法来查询repository去获取已经存在的 execution 。JobExplored 接口提供了这些功能：

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);
    }

上面的代码表示的很明显，JobExplorer是一个只读版的JobRepository，同JobRepository一样，它也能够很容易配置一个工厂类：

之前有提到过，JobRepository 能够配置不同的表前缀用来支持不同的版本或是schema。JobExplorer 也支持同样的特性：

## **1.6.2 JobRegistry** ##

JobRegistry (父接口为 JobLocator )并非强制使用，它能够协助用户在上下文中追踪job是否可用，也能够在应用上下文收集在其他地方(子上下文)创建的job信息。自定义的JobRegistry实现常被用于操作job的名称或是其他属性。框架提供了一个基于map的默认实现，能够从job的名称映射到job的实例：

有两种方法自动注册job进JobRegistry：使用bean的post处理器或是使用注册生命周期组件。这两种机制在下面描述。

### JobRegistryBeanPostProcessor ###

这是post处理器，能够将job在创建时自动注册进JobRegistry：

并不一定要像例子中给post处理器一个id，但是使用id可以在子context中(比如作为作为父 bean 定义)也使用post处理器，这样所有的job在创建时都会自动注册进JobRegistry。

###    **AutomaticJobRegistrar** ###

这是生命周期组件，用于创建子context以及注册这些子context中的job。这种做法有一个好处，虽然job的名字仍然要求全局唯一，但是job的依赖项可以不用全局唯一，它可以有一个“自然”的名字。例如，创建了一组xml配置文件，每个文件有一个job，每个job的ItemReader都有一个相同的名字(如"reader")，如果这些文件被导入到一个上下文中，reader的定义会冲突并且互相覆盖。如果使用了自动注册机就能避免这一切发生。这样集成几个不同的应用模块就变得更容易了：

注册机有两个主要的属性，一个是ApplicationContextFactory数组(这儿创建了一个简单的factory bean)，另一个是jobLoader 。JobLoader 负责管理子context的生命周期以及注册任务到JobRegistry。ApplicationContextFactory 负责创建子 Context，大多数情况下像上面那样使用  
ClassPathXmlApplicationContextFactory。这个工厂类的一个特性是默认情况下他会复制父上下文的一些配置到子上下文。因此如果不变的情况下不需要重新定义子上下文中的 PropertyPlaceholderConfigurer 和AOP配置。  
在必要情况下，AutomaticJobRegistrar 可以和 JobRegistyBeanPostProcessor 一起使用。例如，job有可能既定义在父上下文中也定义在子上下文中的情况。

##    **1.6.3 JobOperator** ##

正如前面所讨论的，JobRepository 提供了对元数据的 CRUD 操作，JobExplorer 提供了对元数据的只读操作。然而，这些操作最常用于联合使用诸多的批量操作类，来对任务进行监测，并完成相当多的任务控制功能，比如停止、重启或对任务进行汇总。在Spring Batch 中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();
    }

上图中展示的操作重现了来自其它接口提供的方法，比如JobLauncher, JobRepository, JobExplorer, 以及 JobRegistry。因为这个原因，所提供的JobOperator的实现SimpleJobOperator的依赖项有很多：

注意 如果你在JobRepository中设置了表前缀，那么不要忘记在JobExplorer中也做同样设置。

## **1.6.4 JobParametersIncrementer** ##

JobOperator 的多数方法都是不言自明的，更多详细的说明可以参见该接口的javadoc(javadoc of the interface)。然而  
startNextInstance方法却有些无所是处。这个方法通常用于启动Job的一个新的实例。但如果 JobExecution 存在若干严重的问题，同时该Job 需要从头重新启动，那么这时候这个方法就相当有用了。不像JobLauncher ，启动新的任务时如果参数不同于任何以往的参数集，这就要求一个新的 JobParameters 对象来触发新的 JobInstance，startNextInstance 方法将使用当前的JobParametersIncrementer绑定到这个任务，并强制其生成新的实例：

public interface JobParametersIncrementer {
         JobParameters getNext(JobParameters parameters);
    }

JobParametersIncrementer 的协议是这样的，当给定一个 JobParameters 对象，它将返回填充了所有可能需要的值 “下一个” JobParameters 对象。这个策略非常有用，因为框架无需知晓变成“下一个”的JobParameters 做了哪些更改。例如，如果任务参数中只包含一个日期参数，那么当创建下一个实例时，这个值就应该是不是该自增一天？或者一周（如果任务是以周为单位运行的话）？任何包含数值类参数的任务，如果需要对其进行区分，都涉及这个问题，如下：

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();
            }
    }

在该示例中，键值“ run.id ”用以区分各个JobInstance。如果当前的JobParameters为空(null)，它将被视为该Job从未运行过，并同时为其初始化，然后返回。反之，非空的时候自增一个数值，再返回。自增的数值可以在命名空间描述中通过Job的“incrementer”属性进行设置：

...
    </job>

## **1.6.5 Stopping a Job** ##

JobOperator 最常见的作用莫过于停止某个Job：

Set<Long> executions = jobOperator.getRunningExecutions("sampleJob");
    jobOperator.stop(executions.iterator().next());

关闭不是立即发生的，因为没有办法将一个任务立刻强制停掉，尤其是当任务进行到开发人员自己的代码段时，框架在此刻是无能为力的，比如某个业务逻辑处理。而一旦控制权还给了框架，它会立刻设置当前 StepExecution 为 BachStatus.STOPPED ，意为停止，然后保存，最后在完成前对JobExecution进行相同的操作。

##    **1.6.6 Aborting a Job** ##

一个job的执行过程当执行到FAILED状态之后，如果它是可重启的，它将会被重启。如果任务的执行过程状态是ABANDONED，那么框架就不会重启它。ABANDONED状态也适用于执行步骤，使得它们可以被跳过，即便是在一个可重启的任务执行之中：如果任务执行过程中碰到在上一次执行失败后标记为ABANDONED的步骤，将会跳过该步骤直接到下一步(这是由任务流定义和执行步骤的退出码决定的)。

如果当前的系统进程死掉了(“kill -9”或系统错误)，job自然也不会运行，但JobRepository是无法侦测到这个错误的，因为进程死掉之前没有对它进行任何通知。你必须手动的告诉它，你知道任务已经失败了还是说考虑放弃这个任务（设置它的状态为FAILED或ABANDONED）-这是业务逻辑层的事情，无法做到自动决策。只有在不可重启的任务中才需要设置为FAILED状态，或者你知道重启后数据还是有效的。Spring Batch Admin中有一系列工具JobService，用以取消正在进行执行的任务。

[20170328164027418]: /images/20220618/a6e216b0743f4a8fb564b633b8bc0a4c.png
[20170328164128889]: /images/20220618/e3199ebab410453d8ff77277c0457044.png
[20170328164154202]: /images/20220618/fc3970922ac74f1ebc54612934b880e0.png
[http_docs.spring.io_spring_docs_3.2.x_spring-framework-reference_html_mvc.html]: http://docs.spring.io/spring/docs/3.2.x/spring-framework-reference/html/mvc.html
[20170328164215703]: /images/20220618/9b2ccc6eb52745eb8979b19dc5d3e389.png
[20170328164233297]: /images/20220618/ce45ea1e6f5c4bee8a9bf68c7d7af2f8.png