Spring Boot集成Spring Batch快速入门Demo

🕗 发布于 2024-07-21 14:17 spring spring boot batch 后端 java

1.什么是Spring Batch？

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

分层架构

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

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

主要概念

2.2.1 JobRepository

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

2.2.2 任务启动器JobLauncher

负责启动任务Job。

2.2.3 任务Job

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

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

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

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

2.2.4 步骤Step

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

通过定义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

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>springboot-demo</artifactId>
        <groupId>com.et</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>SpringBatch</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-batch</artifactId>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

job

第一个简单的任务

package com.et.batch.job;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class FirstJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job firstJob() {
        return jobBuilderFactory.get("firstJob")
                .start(step())
                .build();
    }

    private Step step() {
        return stepBuilderFactory.get("step")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("execute  step....");
                    return RepeatStatus.FINISHED;
                }).build();
    }
}

多步骤的job

package com.et.batch.job;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class MultiStepJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job multiStepJob() {
        /*return jobBuilderFactory.get("multiStepJob")
                .start(step1())
                .next(step2())
                .next(step3())
                .build();*/
        // control the next step by last Status
        return jobBuilderFactory.get("multiStepJob2")
          .start(step1())
                .on(ExitStatus.COMPLETED.getExitCode()).to(step2())
                .from(step2())
                .on(ExitStatus.COMPLETED.getExitCode()).to(step3())
                .from(step3()).end()
                .build();
    }

    private Step step1() {
        return stepBuilderFactory.get("step1")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step1。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step2() {
        return stepBuilderFactory.get("step2")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step2。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step3() {
        return stepBuilderFactory.get("step3")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step3。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }
}

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

package com.et.batch.job;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class FlowJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job flowJob() {
        return jobBuilderFactory.get("flowJob")
                .start(flow())
                .next(step3())
                .end()
                .build();
    }

    private Step step1() {
        return stepBuilderFactory.get("step1")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step1。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step2() {
        return stepBuilderFactory.get("step2")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step2。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step3() {
        return stepBuilderFactory.get("step3")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step3。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Flow flow() {
        return new FlowBuilder<Flow>("flow")
                .start(step1())
                .next(step2())
                .build();
    }
}

并发执行的jobs

package com.et.batch.job;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.stereotype.Component;

@Component
public class SplitJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job splitJob() {
        return jobBuilderFactory.get("splitJob")
                .start(flow1())
                .split(new SimpleAsyncTaskExecutor()).add(flow2())
                .end()
                .build();

    }

    private Step step1() {
        return stepBuilderFactory.get("step1")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step1。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step2() {
        return stepBuilderFactory.get("step2")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step2。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step3() {
        return stepBuilderFactory.get("step3")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step3。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Flow flow1() {
        return new FlowBuilder<Flow>("flow1")
                .start(step1())
                .next(step2())
                .build();
    }

    private Flow flow2() {
        return new FlowBuilder<Flow>("flow2")
                .start(step3())
                .build();
    }
}

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

package com.et.batch.job;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class DeciderJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private MyDecider myDecider;

    @Bean
    public Job deciderJob() {
        return jobBuilderFactory.get("deciderJob")
                .start(step1())
                .next(myDecider)
                .from(myDecider).on("weekend").to(step2())
                .from(myDecider).on("workingDay").to(step3())
                .from(step3()).on("*").to(step4())
                .end()
                .build();
    }

    private Step step1() {
        return stepBuilderFactory.get("step1")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step1。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step2() {
        return stepBuilderFactory.get("step2")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step2。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }

    private Step step3() {
        return stepBuilderFactory.get("step3")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step3。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }


    private Step step4() {
        return stepBuilderFactory.get("step4")
                .tasklet((stepContribution, chunkContext) -> {
                    System.out.println("execute step4。。。");
                    return RepeatStatus.FINISHED;
                }).build();
    }
}

父子嵌套job

package com.et.batch.job;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.JobStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;

@Component
public class NestedJobDemo {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private JobRepository jobRepository;
    @Autowired
    private PlatformTransactionManager platformTransactionManager;

    @Bean
    public Job parentJob() {
        return jobBuilderFactory.get("parentJob")
                .start(childJobOneStep())
                .next(childJobTwoStep())
                .build();
    }


    private Step childJobOneStep() {
        return new JobStepBuilder(new StepBuilder("childJobOneStep"))
                .job(childJobOne())
                .launcher(jobLauncher)
                .repository(jobRepository)
                .transactionManager(platformTransactionManager)
                .build();
    }

    private Step childJobTwoStep() {
        return new JobStepBuilder(new StepBuilder("childJobTwoStep"))
                .job(childJobTwo())
                .launcher(jobLauncher)
                .repository(jobRepository)
                .transactionManager(platformTransactionManager)
                .build();
    }

    private Job childJobOne() {
        return jobBuilderFactory.get("childJobOne")
                .start(
                    stepBuilderFactory.get("childJobOneStep")
                            .tasklet((stepContribution, chunkContext) -> {
                                System.out.println("subtask1。。。");
                                return RepeatStatus.FINISHED;
                            }).build()
                ).build();
    }

    private Job childJobTwo() {
        return jobBuilderFactory.get("childJobTwo")
                .start(
                    stepBuilderFactory.get("childJobTwoStep")
                            .tasklet((stepContribution, chunkContext) -> {
                                System.out.println("subtask2。。。");
                                return RepeatStatus.FINISHED;
                            }).build()
                ).build();
    }
}

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

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

代码仓库

GitHub - Harries/springboot-demo: a simple springboot demo with some components for example: redis,solr,rockmq and so on.

4.测试

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

5.引用

原文地址：https://blog.csdn.net/dot_life/article/details/140574131

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：计算机网络参考模型与5G协议
下一篇：c++树（三）重心

论文《基于现实迷宫地形的电脑鼠设计》深度分析（二）——超声波环境感知算法
《基于现实迷宫地形的电脑鼠设计》是由吴润强、庹忠曜、刘文杰、项璟晨、孙科学等人于2023年发表的一篇优秀期刊论文。其针对现阶段电脑鼠计算量庞大且不适用于现实迷宫地形的问题，特基于超声波测距与传统迷宫算
阅读更多2024-11-18
笔记整理—linux驱动开发部分（13）块设备
整理一下学习的内容，个人观点可能存在错误。本章介绍了什么是块设备，对比了块设备与字符设备的区别，并说明块设备更适合驱动内存类设备的原因。
阅读更多2024-11-18
【Linux】进程状态，优先级，环境变量
进程状态，优先级，环境变量
阅读更多2024-11-18
【Spring】循环引用解决流程，只用一二级缓存？
循环依赖：循环依赖其实就是循环引用，也就是bean互相持有对方，最终形成闭环。比如A依赖于B，B依赖于A循环依赖在spring中是允许存在，spring框架依据三级缓存已经解决了大部分的循环依赖三级缓
阅读更多2024-11-18
Windows、Linux多系统共享蓝牙设备
不需要配对的其实非常不可靠，单独使用可能感受不到，当在很近的距离内使用多套无线鼠标或者无线话筒，就会发现相互干扰的现象，因为它们都是基于一个简单的标签或者信道来区分，而这种划分很容易重复。在网上查资料
阅读更多2024-11-18
自然语言处理技术之细粒度实体识别
细粒度实体识别是自然语言处理（NLP）领域中的一个重要研究方向，其目标是从文本中识别出更加具体和详细的实体类型。相比于传统的实体识别（NER），细粒度实体识别不仅关注常见的实体类别（如。
阅读更多2024-11-18
Linux 域通信
【代码】Linux 域通信。
阅读更多2024-11-18
正则表达式从入门到精通
正则表达式从入门到入土
阅读更多2024-11-18
linux alsa-lib snd_pcm_open函数源码分析（四)
alsa源码分析
阅读更多2024-11-18
算法日记 26-27day 贪心算法
贪心算法
阅读更多2024-11-18