Java Spring Spring Batch

项目搭建

新建一个 Spring Boot 项目，版本为 2.2.4.RELEASE，artifactId 为 spring-batch-start。
然后在 pom.xml 中引入 Spring Batch、MySQL 和 JDBC 依赖，引入后 pom.xml 内容如下所示：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.2.5.RELEASE</version>
  9.         <relativePath/> <!-- lookup parent from repository -->
  10.     </parent>
  11.     <groupId>cc.mrbird</groupId>
  12.     <artifactId>spring-batch-start</artifactId>
  13.     <version>0.0.1-SNAPSHOT</version>
  14.     <name>spring-batch-start</name>
  15.     <description>Demo project for Spring Boot</description>
  17.     <properties>
  18.         <java.version>1.8</java.version>
  19.     </properties>
  21.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.springframework.boot</groupId>
  24.             <artifactId>spring-boot-starter-batch</artifactId>
  25.         </dependency>
  27.         <dependency>
  28.             <groupId>mysql</groupId>
  29.             <artifactId>mysql-connector-java</artifactId>
  30.         </dependency>
  31.         <dependency>
  32.             <groupId>org.springframework.boot</groupId>
  33.             <artifactId>spring-boot-starter-jdbc</artifactId>
  34.         </dependency>
  35.     </dependencies>
  37.     <build>
  38.         <plugins>
  39.             <plugin>
  40.                 <groupId>org.springframework.boot</groupId>
  41.                 <artifactId>spring-boot-maven-plugin</artifactId>
  42.             </plugin>
  43.         </plugins>
  44.     </build>
  45. </project>

在编写代码之前，先来简单了解下 Spring Batch 的组成：
Spring Batch批处理框架实战 - 图1

  • Spring Batch 里最基本的单元就是任务 Job，一个 Job 由若干个步骤 Step 组成。
  • 任务启动器 Job Launcher 负责运行 Job。
  • 任务存储仓库 Job Repository 存储着 Job 的执行状态，参数和日志等信息。Job 处理任务又可以分为三大类：
    • 数据读取 Item Reader
    • 数据中间处理 Item Processor
    • 数据输出 Item Writer。

任务存储仓库可以是关系型数据库 MySQL，非关系型数据库 MongoDB 或者直接存储在内存中，本篇使用的是 MySQL 作为任务存储仓库。
新建一个名称为 springbatch 的 MySQL 数据库，然后导入 org.springframework.batch.core 目录下的 schema-mysql.sql 文件：
Spring Batch批处理框架实战 - 图2Spring Batch批处理框架实战 - 图3
导入后，库表如下图所示：
Spring Batch批处理框架实战 - 图4
然后在项目的配置文件 application.yml 里添加 MySQL 相关配置：

  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.cj.jdbc.Driver
  4.     url: jdbc:mysql://127.0.0.1:3306/springbatch
  5.     username: root
  6.     password: 123456

接着在 Spring Boot 的入口类上添加 @EnableBatchProcessing 注解，表示开启 Spring Batch 批处理功能：

  1. @SpringBootApplication
  2. @EnableBatchProcessing
  3. public class SpringBatchStartApplication {
  4.     public static void main(String[] args) {
  5.         SpringApplication.run(SpringBatchStartApplication.class, args);
  6.     }
  7. }

至此，基本框架搭建好了，下面开始配置一个简单的任务。

编写第一个任务

在 cc.mrbird.batch 目录下新建 job 包，然后在该包下新建一个 FirstJobDemo 类，代码如下所示：

  1. @Component
  2. public class FirstJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  9.     @Bean
  10.     public Job firstJob() {
  11.         return jobBuilderFactory.get("firstJob")
  12.                 .start(step())
  13.                 .build();
  14.     }
  16.     private Step step() {
  17.         return stepBuilderFactory.get("step")
  18.                 .tasklet((contribution, chunkContext) -> {
  19.                     System.out.println("执行步骤....");
  20.                     return RepeatStatus.FINISHED;
  21.                 }).build();
  22.     }
  23. }

上面代码中，注入了JobBuilderFactory任务创建工厂和StepBuilderFactory步骤创建工厂，分别用于创建任务 Job 和步骤 Step。JobBuilderFactory的get方法用于创建一个指定名称的任务，start方法指定任务的开始步骤，步骤通过StepBuilderFactory构建。
步骤 Step 由若干个小任务 Tasklet 组成，所以通过tasklet方法创建。tasklet方法接收一个Tasklet类型参数，Tasklet是一个函数是接口，源码如下：

  1. public interface Tasklet {
  2.     @Nullable
  3.     RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception;
  4. }

所以可以使用 lambda 表达式创建一个匿名实现：

  1. (contribution, chunkContext) -> {
  2.     System.out.println("执行步骤....");
  3.     return RepeatStatus.FINISHED;
  4. }

该匿名实现必须返回一个明确的执行状态，这里返回RepeatStatus.FINISHED表示该小任务执行成功，正常结束。
此外，需要注意的是，配置的任务 Job 必须注册到 Spring IOC 容器中，并且任务的名称和步骤的名称组成唯一。比如上面的例子，任务名称为 firstJob，步骤的名称为 step，如果存在别的任务和步骤组合也叫这个名称的话，则会执行失败。
启动项目，控制台打印日志如下：

  1. ...
  2. 2020-03-06 11:01:11.785  INFO 17324 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=firstJob]] launched with the following parameters: [{}]
  3. 2020-03-06 11:01:11.846  INFO 17324 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step]
  4. 执行步骤....
  5. 2020-03-06 11:01:11.886  INFO 17324 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step] executed in 40ms
  6. 2020-03-06 11:01:11.909  INFO 17324 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=firstJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 101ms

可以看到，任务成功执行了，数据库的库表也将记录相关运行日志。
重新启动项目，控制台并不会再次打印出任务执行日志，因为 Job 名称和 Step 名称组成唯一，执行完的不可重复的任务，不会再次执行。

多步骤任务

一个复杂的任务一般包含多个步骤，下面举个多步骤任务的例子。在 job 包下新建MultiStepJobDemo类：

  1. @Component
  2. public class MultiStepJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  9.     @Bean
  10.     public Job multiStepJob() {
  11.         return jobBuilderFactory.get("multiStepJob")
  12.                 .start(step1())
  13.                 .next(step2())
  14.                 .next(step3())
  15.                 .build();
  16.     }
  18.     private Step step1() {
  19.         return stepBuilderFactory.get("step1")
  20.                 .tasklet((stepContribution, chunkContext) -> {
  21.                     System.out.println("执行步骤一操作。。。");
  22.                     return RepeatStatus.FINISHED;
  23.                 }).build();
  24.     }
  26.     private Step step2() {
  27.         return stepBuilderFactory.get("step2")
  28.                 .tasklet((stepContribution, chunkContext) -> {
  29.                     System.out.println("执行步骤二操作。。。");
  30.                     return RepeatStatus.FINISHED;
  31.                 }).build();
  32.     }
  34.     private Step step3() {
  35.         return stepBuilderFactory.get("step3")
  36.                 .tasklet((stepContribution, chunkContext) -> {
  37.                     System.out.println("执行步骤三操作。。。");
  38.                     return RepeatStatus.FINISHED;
  39.                 }).build();
  40.     }
  41. }

上面代码中，通过step1()、step2()和step3()三个方法创建了三个步骤。Job 里要使用这些步骤，只需要通过JobBuilderFactory的start方法指定第一个步骤，然后通过next方法不断地指定下一个步骤即可。
启动项目，控制台打印日志如下：

  1. 2020-03-06 13:52:52.188  INFO 18472 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=multiStepJob]] launched with the following parameters: [{}]
  2. 2020-03-06 13:52:52.222  INFO 18472 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
  3. 执行步骤一操作。。。
  4. 2020-03-06 13:52:52.251  INFO 18472 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 29ms
  5. 2020-03-06 13:52:52.292  INFO 18472 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step2]
  6. 执行步骤二操作。。。
  7. 2020-03-06 13:52:52.323  INFO 18472 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step2] executed in 30ms
  8. 2020-03-06 13:52:52.375  INFO 18472 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step3]
  9. 执行步骤三操作。。。
  10. 2020-03-06 13:52:52.405  INFO 18472 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step3] executed in 29ms
  11. 2020-03-06 13:52:52.428  INFO 18472 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=multiStepJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 231ms

三个步骤依次执行成功。
多个步骤在执行过程中也可以通过上一个步骤的执行状态来决定是否执行下一个步骤，修改上面的代码：

  1. @Component
  2. public class MultiStepJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  9.     @Bean
  10.     public Job multiStepJob() {
  11.         return jobBuilderFactory.get("multiStepJob2")
  12.                 .start(step1())
  13.                 .on(ExitStatus.COMPLETED.getExitCode()).to(step2())
  14.                 .from(step2())
  15.                 .on(ExitStatus.COMPLETED.getExitCode()).to(step3())
  16.                 .from(step3()).end()
  17.                 .build();
  18.     }
  20.     private Step step1() {
  21.         return stepBuilderFactory.get("step1")
  22.                 .tasklet((stepContribution, chunkContext) -> {
  23.                     System.out.println("执行步骤一操作。。。");
  24.                     return RepeatStatus.FINISHED;
  25.                 }).build();
  26.     }
  28.     private Step step2() {
  29.         return stepBuilderFactory.get("step2")
  30.                 .tasklet((stepContribution, chunkContext) -> {
  31.                     System.out.println("执行步骤二操作。。。");
  32.                     return RepeatStatus.FINISHED;
  33.                 }).build();
  34.     }
  36.     private Step step3() {
  37.         return stepBuilderFactory.get("step3")
  38.                 .tasklet((stepContribution, chunkContext) -> {
  39.                     System.out.println("执行步骤三操作。。。");
  40.                     return RepeatStatus.FINISHED;
  41.                 }).build();
  42.     }
  43. }

multiStepJob()方法的含义是：multiStepJob2 任务先执行 step1，当 step1 状态为完成时，接着执行 step2，当 step2 的状态为完成时，接着执行 step3。ExitStatus.COMPLETED常量表示任务顺利执行完毕，正常退出，该类还包含以下几种退出状态：

  1. public class ExitStatus implements Serializable, Comparable<ExitStatus> {
  3.     /**
  4.      * Convenient constant value representing unknown state - assumed not
  5.      * continuable.
  6.      */
  7.     public static final ExitStatus UNKNOWN = new ExitStatus("UNKNOWN");
  9.     /**
  10.      * Convenient constant value representing continuable state where processing
  11.      * is still taking place, so no further action is required. Used for
  12.      * asynchronous execution scenarios where the processing is happening in
  13.      * another thread or process and the caller is not required to wait for the
  14.      * result.
  15.      */
  16.     public static final ExitStatus EXECUTING = new ExitStatus("EXECUTING");
  18.     /**
  19.      * Convenient constant value representing finished processing.
  20.      */
  21.     public static final ExitStatus COMPLETED = new ExitStatus("COMPLETED");
  23.     /**
  24.      * Convenient constant value representing job that did no processing (e.g.
  25.      * because it was already complete).
  26.      */
  27.     public static final ExitStatus NOOP = new ExitStatus("NOOP");
  29.     /**
  30.      * Convenient constant value representing finished processing with an error.
  31.      */
  32.     public static final ExitStatus FAILED = new ExitStatus("FAILED");
  34.     /**
  35.      * Convenient constant value representing finished processing with
  36.      * interrupted status.
  37.      */
  38.     public static final ExitStatus STOPPED = new ExitStatus("STOPPED");
  40.     ...
  41. }

启动项目，控制台日志打印如下：

  1. 2020-03-06 14:21:49.384  INFO 18745 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=multiStepJob2]] launched with the following parameters: [{}]
  2. 2020-03-06 14:21:49.427  INFO 18745 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
  3. 执行步骤一操作。。。
  4. 2020-03-06 14:21:49.456  INFO 18745 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 29ms
  5. 2020-03-06 14:21:49.501  INFO 18745 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step2]
  6. 执行步骤二操作。。。
  7. 2020-03-06 14:21:49.527  INFO 18745 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step2] executed in 26ms
  8. 2020-03-06 14:21:49.576  INFO 18745 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step3]
  9. 执行步骤三操作。。。
  10. 2020-03-06 14:21:49.604  INFO 18745 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step3] executed in 28ms
  11. 2020-03-06 14:21:49.629  INFO 18745 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=multiStepJob2]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 238ms

Flow 的用法

Flow 的作用就是可以将多个步骤 Step 组合在一起然后再组装到任务 Job 中。举个 Flow 的例子，在 job 包下新建FlowJobDemo类：

  1. @Component
  2. public class FlowJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  9.     @Bean
  10.     public Job flowJob() {
  11.         return jobBuilderFactory.get("flowJob")
  12.                 .start(flow())
  13.                 .next(step3())
  14.                 .end()
  15.                 .build();
  16.     }
  18.     private Step step1() {
  19.         return stepBuilderFactory.get("step1")
  20.                 .tasklet((stepContribution, chunkContext) -> {
  21.                     System.out.println("执行步骤一操作。。。");
  22.                     return RepeatStatus.FINISHED;
  23.                 }).build();
  24.     }
  26.     private Step step2() {
  27.         return stepBuilderFactory.get("step2")
  28.                 .tasklet((stepContribution, chunkContext) -> {
  29.                     System.out.println("执行步骤二操作。。。");
  30.                     return RepeatStatus.FINISHED;
  31.                 }).build();
  32.     }
  34.     private Step step3() {
  35.         return stepBuilderFactory.get("step3")
  36.                 .tasklet((stepContribution, chunkContext) -> {
  37.                     System.out.println("执行步骤三操作。。。");
  38.                     return RepeatStatus.FINISHED;
  39.                 }).build();
  40.     }
  42.     // 创建一个flow对象，包含若干个step
  43.     private Flow flow() {
  44.         return new FlowBuilder<Flow>("flow")
  45.                 .start(step1())
  46.                 .next(step2())
  47.                 .build();
  48.     }
  49. }

上面代码中，通过FlowBuilder将 step1 和 step2 组合在一起，创建了一个名为 flow 的 Flow，然后再将其赋给任务 Job。使用 Flow 和 Step 构建 Job 的区别是，Job 流程中包含 Flow 类型的时候需要在build()方法前调用end()方法。
启动程序，控制台日志打印如下：

  1. 2020-03-06 14:36:42.621  INFO 18865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=flowJob]] launched with the following parameters: [{}]
  2. 2020-03-06 14:36:42.667  INFO 18865 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
  3. 执行步骤一操作。。。
  4. 2020-03-06 14:36:42.697  INFO 18865 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 30ms
  5. 2020-03-06 14:36:42.744  INFO 18865 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step2]
  6. 执行步骤二操作。。。
  7. 2020-03-06 14:36:42.771  INFO 18865 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step2] executed in 27ms
  8. 2020-03-06 14:36:42.824  INFO 18865 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step3]
  9. 执行步骤三操作。。。
  10. 2020-03-06 14:36:42.850  INFO 18865 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step3] executed in 25ms
  11. 2020-03-06 14:36:42.874  INFO 18865 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=flowJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 245ms

并行执行

任务中的步骤除了可以串行执行（一个接着一个执行）外，还可以并行执行，并行执行在特定的业务需求下可以提供任务执行效率。
将任务并行化只需两个简单步骤：

  1. 将步骤 Step 转换为 Flow；
  2. 任务 Job 中指定并行 Flow。

举个例子，在 job 包下新建SplitJobDemo类：

  1. @Component
  2. public class SplitJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  9.     @Bean
  10.     public Job splitJob() {
  11.         return jobBuilderFactory.get("splitJob")
  12.                 .start(flow1())
  13.                 .split(new SimpleAsyncTaskExecutor()).add(flow2())
  14.                 .end()
  15.                 .build();
  17.     }
  19.     private Step step1() {
  20.         return stepBuilderFactory.get("step1")
  21.                 .tasklet((stepContribution, chunkContext) -> {
  22.                     System.out.println("执行步骤一操作。。。");
  23.                     return RepeatStatus.FINISHED;
  24.                 }).build();
  25.     }
  27.     private Step step2() {
  28.         return stepBuilderFactory.get("step2")
  29.                 .tasklet((stepContribution, chunkContext) -> {
  30.                     System.out.println("执行步骤二操作。。。");
  31.                     return RepeatStatus.FINISHED;
  32.                 }).build();
  33.     }
  35.     private Step step3() {
  36.         return stepBuilderFactory.get("step3")
  37.                 .tasklet((stepContribution, chunkContext) -> {
  38.                     System.out.println("执行步骤三操作。。。");
  39.                     return RepeatStatus.FINISHED;
  40.                 }).build();
  41.     }
  43.     private Flow flow1() {
  44.         return new FlowBuilder<Flow>("flow1")
  45.                 .start(step1())
  46.                 .next(step2())
  47.                 .build();
  48.     }
  50.     private Flow flow2() {
  51.         return new FlowBuilder<Flow>("flow2")
  52.                 .start(step3())
  53.                 .build();
  54.     }
  55. }

上面例子中，创建了两个 Flow：flow1（包含 step1 和 step2）和 flow2（包含 step3）。然后通过JobBuilderFactory的split方法，指定一个异步执行器，将 flow1 和 flow2 异步执行（也就是并行）。
启动项目，控制台日志打印如下：

  1. 2020-03-06 15:25:43.602  INFO 19449 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=splitJob]] launched with the following parameters: [{}]
  2. 2020-03-06 15:25:43.643  INFO 19449 --- [cTaskExecutor-1] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step3]
  3. 2020-03-06 15:25:43.650  INFO 19449 --- [cTaskExecutor-2] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
  4. 执行步骤三操作。。。
  5. 执行步骤一操作。。。
  6. 2020-03-06 15:25:43.673  INFO 19449 --- [cTaskExecutor-2] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 23ms
  7. 2020-03-06 15:25:43.674  INFO 19449 --- [cTaskExecutor-1] o.s.batch.core.step.AbstractStep         : Step: [step3] executed in 31ms
  8. 2020-03-06 15:25:43.714  INFO 19449 --- [cTaskExecutor-2] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step2]
  9. 执行步骤二操作。。。
  10. 2020-03-06 15:25:43.738  INFO 19449 --- [cTaskExecutor-2] o.s.batch.core.step.AbstractStep         : Step: [step2] executed in 24ms
  11. 2020-03-06 15:25:43.758  INFO 19449 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=splitJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 146ms

可以看到 step3 并没有在 step2 后才执行，说明步骤已经是并行化的（开启并行化后，并行的步骤执行顺序并不能 100%确定，因为线程调度具有不确定性）。

任务决策器

决策器的作用就是可以指定程序在不同的情况下运行不同的任务流程，比如今天是周末，则让任务执行 step1 和 step2，如果是工作日，则之心 step1 和 step3。
使用决策器前，需要自定义一个决策器的实现。在 cc.mrbird.batch 包下新建 decider 包，然后创建MyDecider类，实现JobExecutionDecider接口：

  1. @Component
  2. public class MyDecider implements JobExecutionDecider {
  3.     @Override
  4.     public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
  5.         LocalDate now = LocalDate.now();
  6.         DayOfWeek dayOfWeek = now.getDayOfWeek();
  8.         if (dayOfWeek == DayOfWeek.SATURDAY || dayOfWeek == DayOfWeek.SUNDAY) {
  9.             return new FlowExecutionStatus("weekend");
  10.         } else {
  11.             return new FlowExecutionStatus("workingDay");
  12.         }
  13.     }
  14. }

MyDecider实现JobExecutionDecider接口的decide方法，该方法返回FlowExecutionStatus。上面的逻辑是：判断今天是否是周末，如果是，返回FlowExecutionStatus("weekend")状态，否则返回FlowExecutionStatus("workingDay")状态。
下面演示如何在任务 Job 里使用决策器。在 job 包下新建DeciderJobDemo：

  1. @Component
  2. public class DeciderJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  8.     @Autowired
  9.     private MyDecider myDecider;
  11.     @Bean
  12.     public Job deciderJob() {
  13.         return jobBuilderFactory.get("deciderJob")
  14.             .start(step1())
  15.             .next(myDecider)
  16.             .from(myDecider).on("weekend").to(step2())
  17.             .from(myDecider).on("workingDay").to(step3())
  18.             .from(step3()).on("*").to(step4())
  19.             .end()
  20.             .build();
  21.     }
  23.     private Step step1() {
  24.         return stepBuilderFactory.get("step1")
  25.             .tasklet((stepContribution, chunkContext) -> {
  26.                 System.out.println("执行步骤一操作。。。");
  27.                 return RepeatStatus.FINISHED;
  28.             }).build();
  29.     }
  31.     private Step step2() {
  32.         return stepBuilderFactory.get("step2")
  33.             .tasklet((stepContribution, chunkContext) -> {
  34.                 System.out.println("执行步骤二操作。。。");
  35.                 return RepeatStatus.FINISHED;
  36.             }).build();
  37.     }
  39.     private Step step3() {
  40.         return stepBuilderFactory.get("step3")
  41.             .tasklet((stepContribution, chunkContext) -> {
  42.                 System.out.println("执行步骤三操作。。。");
  43.                 return RepeatStatus.FINISHED;
  44.             }).build();
  45.     }
  48.     private Step step4() {
  49.         return stepBuilderFactory.get("step4")
  50.             .tasklet((stepContribution, chunkContext) -> {
  51.                 System.out.println("执行步骤四操作。。。");
  52.                 return RepeatStatus.FINISHED;
  53.             }).build();
  54.     }
  55. }

上面代码中，注入了自定义决策器MyDecider，然后在jobDecider()方法里使用了该决策器：

  1. @Bean
  2. public Job deciderJob() {
  3.     return jobBuilderFactory.get("deciderJob")
  4.         .start(step1())
  5.         .next(myDecider)
  6.         .from(myDecider).on("weekend").to(step2())
  7.         .from(myDecider).on("workingDay").to(step3())
  8.         .from(step3()).on("*").to(step4())
  9.         .end()
  10.         .build();
  11. }

这段代码的含义是：任务 deciderJob 首先执行 step1，然后指定自定义决策器，如果决策器返回 weekend，那么执行 step2，如果决策器返回 workingDay，那么执行 step3。如果执行了 step3，那么无论 step3 的结果是什么，都将执行 step4。
启动项目，控制台输出如下所示：

  1. 2020-03-06 16:09:10.541  INFO 19873 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=deciderJob]] launched with the following parameters: [{}]
  2. 2020-03-06 16:09:10.609  INFO 19873 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
  3. 执行步骤一操作。。。
  4. 2020-03-06 16:09:10.641  INFO 19873 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 32ms
  5. 2020-03-06 16:09:10.692  INFO 19873 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step3]
  6. 执行步骤三操作。。。
  7. 2020-03-06 16:09:10.723  INFO 19873 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step3] executed in 31ms
  8. 2020-03-06 16:09:10.769  INFO 19873 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step4]
  9. 执行步骤四操作。。。
  10. 2020-03-06 16:09:10.797  INFO 19873 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step4] executed in 27ms
  11. 2020-03-06 16:09:10.818  INFO 19873 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=deciderJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 256ms

因为今天是 2020 年 03 月 06 日星期五，是工作日，所以任务执行了 step1、step3 和 step4。

任务嵌套

任务 Job 除了可以由 Step 或者 Flow 构成外，还可以将多个任务 Job 转换为特殊的 Step，然后再赋给另一个任务 Job，这就是任务的嵌套。
举个例子，在 job 包下新建NestedJobDemo类：

  1. @Component
  2. public class NestedJobDemo {
  4.     @Autowired
  5.     private JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     private StepBuilderFactory stepBuilderFactory;
  8.     @Autowired
  9.     private JobLauncher jobLauncher;
  10.     @Autowired
  11.     private JobRepository jobRepository;
  12.     @Autowired
  13.     private PlatformTransactionManager platformTransactionManager;
  15.     // 父任务
  16.     @Bean
  17.     public Job parentJob() {
  18.         return jobBuilderFactory.get("parentJob")
  19.             .start(childJobOneStep())
  20.             .next(childJobTwoStep())
  21.             .build();
  22.     }
  25.     // 将任务转换为特殊的步骤
  26.     private Step childJobOneStep() {
  27.         return new JobStepBuilder(new StepBuilder("childJobOneStep"))
  28.             .job(childJobOne())
  29.             .launcher(jobLauncher)
  30.             .repository(jobRepository)
  31.             .transactionManager(platformTransactionManager)
  32.             .build();
  33.     }
  35.     // 将任务转换为特殊的步骤
  36.     private Step childJobTwoStep() {
  37.         return new JobStepBuilder(new StepBuilder("childJobTwoStep"))
  38.             .job(childJobTwo())
  39.             .launcher(jobLauncher)
  40.             .repository(jobRepository)
  41.             .transactionManager(platformTransactionManager)
  42.             .build();
  43.     }
  45.     // 子任务一
  46.     private Job childJobOne() {
  47.         return jobBuilderFactory.get("childJobOne")
  48.             .start(
  49.             stepBuilderFactory.get("childJobOneStep")
  50.             .tasklet((stepContribution, chunkContext) -> {
  51.                 System.out.println("子任务一执行步骤。。。");
  52.                 return RepeatStatus.FINISHED;
  53.             }).build()
  54.         ).build();
  55.     }
  57.     // 子任务二
  58.     private Job childJobTwo() {
  59.         return jobBuilderFactory.get("childJobTwo")
  60.             .start(
  61.             stepBuilderFactory.get("childJobTwoStep")
  62.             .tasklet((stepContribution, chunkContext) -> {
  63.                 System.out.println("子任务二执行步骤。。。");
  64.                 return RepeatStatus.FINISHED;
  65.             }).build()
  66.         ).build();
  67.     }
  68. }

上面代码中，通过childJobOne()和childJobTwo()方法创建了两个任务 Job，这里没什么好说的，前面都介绍过。关键在于childJobOneStep()方法和childJobTwoStep()方法。在childJobOneStep()方法中，通过JobStepBuilder构建了一个名称为childJobOneStep的 Step，顾名思义，它是一个任务型 Step 的构造工厂，可以将任务转换为“特殊”的步骤。在构建过程中，还需要传入任务执行器 JobLauncher、任务仓库 JobRepository 和事务管理器 PlatformTransactionManager。
将任务转换为特殊的步骤后，将其赋给父任务 parentJob 即可，流程和前面介绍的一致。
配置好后，启动项目，控制台输出如下所示：

  1. 2020-03-06 16:58:39.771  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=parentJob]] launched with the following parameters: [{}]
  2. 2020-03-06 16:58:39.812  INFO 21588 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [childJobOneStep]
  3. 2020-03-06 16:58:39.866  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=childJobOne]] launched with the following parameters: [{}]
  4. 2020-03-06 16:58:39.908  INFO 21588 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [childJobOneStep]
  5. 子任务一执行步骤。。。
  6. 2020-03-06 16:58:39.940  INFO 21588 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [childJobOneStep] executed in 32ms
  7. 2020-03-06 16:58:39.960  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=childJobOne]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 86ms
  8. 2020-03-06 16:58:39.983  INFO 21588 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [childJobOneStep] executed in 171ms
  9. 2020-03-06 16:58:40.019  INFO 21588 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [childJobTwoStep]
  10. 2020-03-06 16:58:40.067  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=childJobTwo]] launched with the following parameters: [{}]
  11. 2020-03-06 16:58:40.102  INFO 21588 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [childJobTwoStep]
  12. 子任务二执行步骤。。。
  13. 2020-03-06 16:58:40.130  INFO 21588 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [childJobTwoStep] executed in 28ms
  14. 2020-03-06 16:58:40.152  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=childJobTwo]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 75ms
  15. 2020-03-06 16:58:40.157  INFO 21588 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [childJobTwoStep] executed in 138ms
  16. 2020-03-06 16:58:40.177  INFO 21588 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=parentJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 398ms

参考资料

Spring Batch 官方文档: https://docs.spring.io/spring-batch/docs/4.3.x/reference/html/spring-batch-intro.html#spring-batch-intro