devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spring Boot集成Spring Batch快速入门Demo_springboot + springbatch

作者：我家小花儿 | 2024-07-31 23:56:16

赞

踩

springboot + springbatch

1.什么是Spring Batch？

Spring Batch 是一个轻量级的开源框架，它提供了一种简单的方式来处理大量的数据。它基于Spring框架，提供了一套批处理框架，可以处理各种类型的批处理任务，如ETL、数据导入/导出、报表生成等。Spring Batch提供了一些重要的概念，如Job、Step、ItemReader、ItemProcessor、ItemWriter等，这些概念可以帮助我们构建可重用的批处理应用程序。通过Spring Batch，我们可以轻松地实现批处理的并发、容错、重试等功能，同时也可以方便地与其他Spring组件集成，如Spring Boot、Spring Data等。总之，Spring Batch是一个非常强大、灵活、易于使用的批处理框架，可以帮助我们快速构建高效、可靠的批处理应用程序。

分层架构

spring-batch-layers

可以看到它分为三层，分别是：

  • Application应用层：包含了所有任务batch jobs和开发人员自定义的代码，主要是根据项目需要开发的业务流程等。
  • Batch Core核心层：包含启动和管理任务的运行环境类，如JobLauncher等。
  • Batch Infrastructure基础层：上面两层是建立在基础层之上的，包含基础的读入reader和写出writer、重试框架等。

主要概念

spring-batch-reference-model

2.2.1 JobRepository

专门负责与数据库打交道，对整个批处理的新增、更新、执行进行记录。所以Spring Batch是需要依赖数据库来管理的。

2.2.2 任务启动器JobLauncher

负责启动任务Job。

2.2.3 任务Job

Job是封装整个批处理过程的单位，跑一个批处理任务，就是跑一个Job所定义的内容。

111

  上图介绍了Job的一些相关概念：

  • Job：封装处理实体，定义过程逻辑。
  • JobInstance：Job的运行实例，不同的实例，参数不同，所以定义好一个Job后可以通过不同参数运行多次。
  • JobParameters：与JobInstance相关联的参数。
  • JobExecution：代表Job的一次实际执行，可能成功、可能失败。

所以，开发人员要做的事情，就是定义Job。

2.2.4 步骤Step

Step是对Job某个过程的封装，一个Job可以包含一个或多个Step，一步步的Step按特定逻辑执行，才代表Job执行完成。

222

  通过定义Step来组装Job可以更灵活地实现复杂的业务逻辑。

2.2.5 输入——处理——输出

所以，定义一个Job关键是定义好一个或多个Step，然后把它们组装好即可。而定义Step有多种方法，但有一种常用的模型就是输入——处理——输出，即Item Reader、Item Processor和Item Writer。比如通过Item Reader从文件输入数据，然后通过Item Processor进行业务处理和数据转换，最后通过Item Writer写到数据库中去。 Spring Batch为我们提供了许多开箱即用的Reader和Writer，非常方便。

 

2.环境搭建

参照代码仓库mysql模块里面docker目录搭建

3.代码工程

实验目标

如何使用 Spring Boot 创建各种不同类型 Spring Batch Job

pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>springboot-demo</artifactId>
  7.         <groupId>com.et</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  11.  
  12.     <artifactId>SpringBatch</artifactId>
  13.  
  14.     <properties>
  15.         <maven.compiler.source>8</maven.compiler.source>
  16.         <maven.compiler.target>8</maven.compiler.target>
  17.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-web</artifactId>
  22.         </dependency>
  23.  
  24.         <dependency>
  25.             <groupId>org.springframework.boot</groupId>
  26.             <artifactId>spring-boot-autoconfigure</artifactId>
  27.         </dependency>
  28.         <dependency>
  29.             <groupId>org.springframework.boot</groupId>
  30.             <artifactId>spring-boot-starter-test</artifactId>
  31.             <scope>test</scope>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>org.springframework.boot</groupId>
  35.             <artifactId>spring-boot-starter-batch</artifactId>
  36.         </dependency>
  37.  
  38.         <dependency>
  39.             <groupId>mysql</groupId>
  40.             <artifactId>mysql-connector-java</artifactId>
  41.         </dependency>
  42.         <dependency>
  43.             <groupId>org.springframework.boot</groupId>
  44.             <artifactId>spring-boot-starter-jdbc</artifactId>
  45.         </dependency>
  46.     </dependencies>
  47.     <build>
  48.         <plugins>
  49.             <plugin>
  50.                 <groupId>org.springframework.boot</groupId>
  51.                 <artifactId>spring-boot-maven-plugin</artifactId>
  52.             </plugin>
  53.         </plugins>
  54.     </build>
  55. </project>

job

第一个简单的任务

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.Step;
  5. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  6. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  7. import org.springframework.batch.repeat.RepeatStatus;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.stereotype.Component;
  11.  
  12. @Component
  13. public class FirstJobDemo {
  14.  
  15.     @Autowired
  16.     private JobBuilderFactory jobBuilderFactory;
  17.     @Autowired
  18.     private StepBuilderFactory stepBuilderFactory;
  19.  
  20.     @Bean
  21.     public Job firstJob() {
  22.         return jobBuilderFactory.get("firstJob")
  23.                 .start(step())
  24.                 .build();
  25.     }
  26.  
  27.     private Step step() {
  28.         return stepBuilderFactory.get("step")
  29.                 .tasklet((contribution, chunkContext) -> {
  30.                     System.out.println("execute  step....");
  31.                     return RepeatStatus.FINISHED;
  32.                 }).build();
  33.     }
  34. }

多步骤的job

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.ExitStatus;
  4. import org.springframework.batch.core.Job;
  5. import org.springframework.batch.core.Step;
  6. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  7. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  8. import org.springframework.batch.repeat.RepeatStatus;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.context.annotation.Bean;
  11. import org.springframework.stereotype.Component;
  12.  
  13. @Component
  14. public class MultiStepJobDemo {
  15.  
  16.     @Autowired
  17.     private JobBuilderFactory jobBuilderFactory;
  18.     @Autowired
  19.     private StepBuilderFactory stepBuilderFactory;
  20.  
  21.     @Bean
  22.     public Job multiStepJob() {
  23.         /*return jobBuilderFactory.get("multiStepJob")
  24.                 .start(step1())
  25.                 .next(step2())
  26.                 .next(step3())
  27.                 .build();*/
  28.         // control the next step by last Status
  29.         return jobBuilderFactory.get("multiStepJob2")
  30.           .start(step1())
  31.                 .on(ExitStatus.COMPLETED.getExitCode()).to(step2())
  32.                 .from(step2())
  33.                 .on(ExitStatus.COMPLETED.getExitCode()).to(step3())
  34.                 .from(step3()).end()
  35.                 .build();
  36.     }
  37.  
  38.     private Step step1() {
  39.         return stepBuilderFactory.get("step1")
  40.                 .tasklet((stepContribution, chunkContext) -> {
  41.                     System.out.println("execute step1。。。");
  42.                     return RepeatStatus.FINISHED;
  43.                 }).build();
  44.     }
  45.  
  46.     private Step step2() {
  47.         return stepBuilderFactory.get("step2")
  48.                 .tasklet((stepContribution, chunkContext) -> {
  49.                     System.out.println("execute step2。。。");
  50.                     return RepeatStatus.FINISHED;
  51.                 }).build();
  52.     }
  53.  
  54.     private Step step3() {
  55.         return stepBuilderFactory.get("step3")
  56.                 .tasklet((stepContribution, chunkContext) -> {
  57.                     System.out.println("execute step3。。。");
  58.                     return RepeatStatus.FINISHED;
  59.                 }).build();
  60.     }
  61. }

多flow控制的job， 创建一个flow对象，包含若干个step

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.Step;
  5. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  6. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  7. import org.springframework.batch.core.job.builder.FlowBuilder;
  8. import org.springframework.batch.core.job.flow.Flow;
  9. import org.springframework.batch.repeat.RepeatStatus;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.context.annotation.Bean;
  12. import org.springframework.stereotype.Component;
  13.  
  14. @Component
  15. public class FlowJobDemo {
  16.  
  17.     @Autowired
  18.     private JobBuilderFactory jobBuilderFactory;
  19.     @Autowired
  20.     private StepBuilderFactory stepBuilderFactory;
  21.  
  22.     @Bean
  23.     public Job flowJob() {
  24.         return jobBuilderFactory.get("flowJob")
  25.                 .start(flow())
  26.                 .next(step3())
  27.                 .end()
  28.                 .build();
  29.     }
  30.  
  31.     private Step step1() {
  32.         return stepBuilderFactory.get("step1")
  33.                 .tasklet((stepContribution, chunkContext) -> {
  34.                     System.out.println("execute step1。。。");
  35.                     return RepeatStatus.FINISHED;
  36.                 }).build();
  37.     }
  38.  
  39.     private Step step2() {
  40.         return stepBuilderFactory.get("step2")
  41.                 .tasklet((stepContribution, chunkContext) -> {
  42.                     System.out.println("execute step2。。。");
  43.                     return RepeatStatus.FINISHED;
  44.                 }).build();
  45.     }
  46.  
  47.     private Step step3() {
  48.         return stepBuilderFactory.get("step3")
  49.                 .tasklet((stepContribution, chunkContext) -> {
  50.                     System.out.println("execute step3。。。");
  51.                     return RepeatStatus.FINISHED;
  52.                 }).build();
  53.     }
  54.  
  55.     private Flow flow() {
  56.         return new FlowBuilder<Flow>("flow")
  57.                 .start(step1())
  58.                 .next(step2())
  59.                 .build();
  60.     }
  61. }

并发执行的jobs

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.Step;
  5. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  6. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  7. import org.springframework.batch.core.job.builder.FlowBuilder;
  8. import org.springframework.batch.core.job.flow.Flow;
  9. import org.springframework.batch.repeat.RepeatStatus;
  10. import org.springframework.beans.factory.annotation.Autowired;
  11. import org.springframework.context.annotation.Bean;
  12. import org.springframework.core.task.SimpleAsyncTaskExecutor;
  13. import org.springframework.stereotype.Component;
  14.  
  15. @Component
  16. public class SplitJobDemo {
  17.  
  18.     @Autowired
  19.     private JobBuilderFactory jobBuilderFactory;
  20.     @Autowired
  21.     private StepBuilderFactory stepBuilderFactory;
  22.  
  23.     @Bean
  24.     public Job splitJob() {
  25.         return jobBuilderFactory.get("splitJob")
  26.                 .start(flow1())
  27.                 .split(new SimpleAsyncTaskExecutor()).add(flow2())
  28.                 .end()
  29.                 .build();
  30.  
  31.     }
  32.  
  33.     private Step step1() {
  34.         return stepBuilderFactory.get("step1")
  35.                 .tasklet((stepContribution, chunkContext) -> {
  36.                     System.out.println("execute step1。。。");
  37.                     return RepeatStatus.FINISHED;
  38.                 }).build();
  39.     }
  40.  
  41.     private Step step2() {
  42.         return stepBuilderFactory.get("step2")
  43.                 .tasklet((stepContribution, chunkContext) -> {
  44.                     System.out.println("execute step2。。。");
  45.                     return RepeatStatus.FINISHED;
  46.                 }).build();
  47.     }
  48.  
  49.     private Step step3() {
  50.         return stepBuilderFactory.get("step3")
  51.                 .tasklet((stepContribution, chunkContext) -> {
  52.                     System.out.println("execute step3。。。");
  53.                     return RepeatStatus.FINISHED;
  54.                 }).build();
  55.     }
  56.  
  57.     private Flow flow1() {
  58.         return new FlowBuilder<Flow>("flow1")
  59.                 .start(step1())
  60.                 .next(step2())
  61.                 .build();
  62.     }
  63.  
  64.     private Flow flow2() {
  65.         return new FlowBuilder<Flow>("flow2")
  66.                 .start(step3())
  67.                 .build();
  68.     }
  69. }

根据上次运行结果判断是否执行下一步

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.Step;
  5. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  6. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  7. import org.springframework.batch.repeat.RepeatStatus;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.context.annotation.Bean;
  10. import org.springframework.stereotype.Component;
  11.  
  12. @Component
  13. public class DeciderJobDemo {
  14.  
  15.     @Autowired
  16.     private JobBuilderFactory jobBuilderFactory;
  17.     @Autowired
  18.     private StepBuilderFactory stepBuilderFactory;
  19.     @Autowired
  20.     private MyDecider myDecider;
  21.  
  22.     @Bean
  23.     public Job deciderJob() {
  24.         return jobBuilderFactory.get("deciderJob")
  25.                 .start(step1())
  26.                 .next(myDecider)
  27.                 .from(myDecider).on("weekend").to(step2())
  28.                 .from(myDecider).on("workingDay").to(step3())
  29.                 .from(step3()).on("*").to(step4())
  30.                 .end()
  31.                 .build();
  32.     }
  33.  
  34.     private Step step1() {
  35.         return stepBuilderFactory.get("step1")
  36.                 .tasklet((stepContribution, chunkContext) -> {
  37.                     System.out.println("execute step1。。。");
  38.                     return RepeatStatus.FINISHED;
  39.                 }).build();
  40.     }
  41.  
  42.     private Step step2() {
  43.         return stepBuilderFactory.get("step2")
  44.                 .tasklet((stepContribution, chunkContext) -> {
  45.                     System.out.println("execute step2。。。");
  46.                     return RepeatStatus.FINISHED;
  47.                 }).build();
  48.     }
  49.  
  50.     private Step step3() {
  51.         return stepBuilderFactory.get("step3")
  52.                 .tasklet((stepContribution, chunkContext) -> {
  53.                     System.out.println("execute step3。。。");
  54.                     return RepeatStatus.FINISHED;
  55.                 }).build();
  56.     }
  57.  
  58.  
  59.     private Step step4() {
  60.         return stepBuilderFactory.get("step4")
  61.                 .tasklet((stepContribution, chunkContext) -> {
  62.                     System.out.println("execute step4。。。");
  63.                     return RepeatStatus.FINISHED;
  64.                 }).build();
  65.     }
  66. }

父子嵌套job

  1. package com.et.batch.job;
  2.  
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.Step;
  5. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  6. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  7. import org.springframework.batch.core.launch.JobLauncher;
  8. import org.springframework.batch.core.repository.JobRepository;
  9. import org.springframework.batch.core.step.builder.JobStepBuilder;
  10. import org.springframework.batch.core.step.builder.StepBuilder;
  11. import org.springframework.batch.repeat.RepeatStatus;
  12. import org.springframework.beans.factory.annotation.Autowired;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.stereotype.Component;
  15. import org.springframework.transaction.PlatformTransactionManager;
  16.  
  17. @Component
  18. public class NestedJobDemo {
  19.  
  20.     @Autowired
  21.     private JobBuilderFactory jobBuilderFactory;
  22.     @Autowired
  23.     private StepBuilderFactory stepBuilderFactory;
  24.     @Autowired
  25.     private JobLauncher jobLauncher;
  26.     @Autowired
  27.     private JobRepository jobRepository;
  28.     @Autowired
  29.     private PlatformTransactionManager platformTransactionManager;
  30.  
  31.     @Bean
  32.     public Job parentJob() {
  33.         return jobBuilderFactory.get("parentJob")
  34.                 .start(childJobOneStep())
  35.                 .next(childJobTwoStep())
  36.                 .build();
  37.     }
  38.  
  39.  
  40.     private Step childJobOneStep() {
  41.         return new JobStepBuilder(new StepBuilder("childJobOneStep"))
  42.                 .job(childJobOne())
  43.                 .launcher(jobLauncher)
  44.                 .repository(jobRepository)
  45.                 .transactionManager(platformTransactionManager)
  46.                 .build();
  47.     }
  48.  
  49.     private Step childJobTwoStep() {
  50.         return new JobStepBuilder(new StepBuilder("childJobTwoStep"))
  51.                 .job(childJobTwo())
  52.                 .launcher(jobLauncher)
  53.                 .repository(jobRepository)
  54.                 .transactionManager(platformTransactionManager)
  55.                 .build();
  56.     }
  57.  
  58.     private Job childJobOne() {
  59.         return jobBuilderFactory.get("childJobOne")
  60.                 .start(
  61.                     stepBuilderFactory.get("childJobOneStep")
  62.                             .tasklet((stepContribution, chunkContext) -> {
  63.                                 System.out.println("subtask1。。。");
  64.                                 return RepeatStatus.FINISHED;
  65.                             }).build()
  66.                 ).build();
  67.     }
  68.  
  69.     private Job childJobTwo() {
  70.         return jobBuilderFactory.get("childJobTwo")
  71.                 .start(
  72.                     stepBuilderFactory.get("childJobTwoStep")
  73.                             .tasklet((stepContribution, chunkContext) -> {
  74.                                 System.out.println("subtask2。。。");
  75.                                 return RepeatStatus.FINISHED;
  76.                             }).build()
  77.                 ).build();
  78.     }
  79. }

application.yaml

自动会初始化脚本，只需要建立以恶搞空库就行

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/springbatch
    username: root
    password: 123456

  batch:
    jdbc:
      schema: classpath:org/springframework/batch/core/schema-mysql.sql
    initialize-schema: always #Since Spring Boot 2.5.0 use spring.batch.jdbc.initialize-schema=never

    job:
      enabled: true

以上只是一些关键代码，所有代码请参见下面代码仓库

代码仓库

4.测试

  • 启动Spring Boot应用程序，系统会自动运行job，跑过一次，下次启动不会继续执行
  • 如果要执行定时任务，可以利用spring提供的scheduledTaskRegistrar注册一个定时任务，扫描最新的定时任务，将这些定时任务注册到scheduleFuture中从而实现动态定时任务。

5.引用

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/911351
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号