devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Spring Batch快速入门_springbatch教程

作者：从前慢现在也慢 | 2024-03-08 15:00:58

赞

踩

springbatch教程

一、Spring Batch简介

在企业开发中可能会面临到一些需要处理较大数据量的场景，例如将一个表的全部数据导入到另一张表结构类似的表中、批量读取一个或多个文件内容并写入到数据库中，又或者将一张表的数据批量更新到另一张表中。Spring Batch可以快速的开发这种场景下的批处理应用程序。
Spring Batch提供了在处理大量数据时必不可少的可重用功能，包括 日志记录/跟踪、事务管理、作业处理统计信息、作业重新启动、跳过和资源管理。对于 大数据量和高性能的批处理任务，Spring Batch 同样提供了高级功能和特性来支持，例如 分区功能、远程功能等，大大简化了批处理应用的开发，将开发人员从复杂的任务配置管理过程中解放出来，让我们可以更多地去关注核心的业务的处理过程。总之，通过 Spring Batch 我们就能够实现简单的或者复杂的和大数据量的批处理作业。

1.1 Spring Batch结构图

  • JobRepository：用来注册job的容器

  • JobLauncher：用来启动Job的接口

  • Job：实际执行的任务，包含一个或多个Step

  • Step：包含ItemReader、ItemProcessor和ItemWriter

  • ItemReader：用来读取数据的接口

  • ItemProcessor：用来处理数据的接口

  • ItemWriter： 用来输出数据的接口

官网地址如下：

1.2 处理架构

任务的处理是在 Step 这个阶段定义的。在 Step 中，需要定义数据的读取、数据的处理、数据的写出操作，在这三个阶段中，数据的处理是真正进行数据处理的地方。具体 Step 的流程如下图所示：

  • Reader（架构图中的 Item Reader）：主要的任务是定义数据的读取操作，包括读取文件的位置、对读取首先要进行的划分（如以 ',' 作为分隔符）、将读取到的文件映射到相关对象的属性字段等

  • Process（架构图中的 Item Processor）：这里是真正对数据进行处理的地方，数据的处理逻辑都在这里定义

  • Writer（架构图中的 Item Writer）：这个阶段的主要任务是定义数据的输出操作，包括将数据写入到数据库等

1.2 Spring Batch数据库表格

当批处理作业需要操作数据库时，Spring Batch要求在数据库中创建好批处理作业的元数据的存储表格。如下，其中以batch开头的表，是Spring Batch用来存储每次执行作业所产生的元数据。而student表则是作为我们这个Demo中数据的来源：

下图显示了所有6张表的ERD模型及其相互关系（摘自官网）：

1.3 Spring Batch Sql脚本

在数据库中执行官方元数据模式SQL脚本：

  1. -- do not edit this file
  2. -- BATCH JOB 实例表 包含与aJobInstance相关的所有信息
  3. -- JOB ID由batch_job_seq分配
  4. -- JOB 名称,与spring配置一致
  5. -- JOB KEY 对job参数的MD5编码,正因为有这个字段的存在，同一个job如果第一次运行成功，第二次再运行会抛出JobInstanceAlreadyCompleteException异常。
  6. CREATE TABLE BATCH_JOB_INSTANCE  (
  7.     JOB_INSTANCE_ID BIGINT  NOT NULL PRIMARY KEY ,
  8.     VERSION BIGINT ,
  9.     JOB_NAME VARCHAR(100) NOT NULL,
  10.     JOB_KEY VARCHAR(32) NOT NULL,
  11.     constraint JOB_INST_UN unique (JOB_NAME, JOB_KEY)
  12. ) ENGINE=InnoDB;
  13.  
  14. -- 该BATCH_JOB_EXECUTION表包含与该JobExecution对象相关的所有信息
  15. CREATE TABLE BATCH_JOB_EXECUTION  (
  16.     JOB_EXECUTION_ID BIGINT  NOT NULL PRIMARY KEY ,
  17.     VERSION BIGINT  ,
  18.     JOB_INSTANCE_ID BIGINT NOT NULL,
  19.     CREATE_TIME DATETIME NOT NULL,
  20.     START_TIME DATETIME DEFAULT NULL ,
  21.     END_TIME DATETIME DEFAULT NULL ,
  22.     STATUS VARCHAR(10) ,
  23.     EXIT_CODE VARCHAR(2500) ,
  24.     EXIT_MESSAGE VARCHAR(2500) ,
  25.     LAST_UPDATED DATETIME,
  26.     JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
  27.     constraint JOB_INST_EXEC_FK foreign key (JOB_INSTANCE_ID)
  28.     references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
  29. ) ENGINE=InnoDB;
  30.  
  31. -- 该表包含与该JobParameters对象相关的所有信息
  32. CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
  33.     JOB_EXECUTION_ID BIGINT NOT NULL ,
  34.     TYPE_CD VARCHAR(6) NOT NULL ,
  35.     KEY_NAME VARCHAR(100) NOT NULL ,
  36.     STRING_VAL VARCHAR(250) ,
  37.     DATE_VAL DATETIME DEFAULT NULL ,
  38.     LONG_VAL BIGINT ,
  39.     DOUBLE_VAL DOUBLE PRECISION ,
  40.     IDENTIFYING CHAR(1) NOT NULL ,
  41.     constraint JOB_EXEC_PARAMS_FK foreign key (JOB_EXECUTION_ID)
  42.     references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
  43. ) ENGINE=InnoDB;
  44.  
  45. -- 该表包含与该StepExecution 对象相关的所有信息
  46. CREATE TABLE BATCH_STEP_EXECUTION  (
  47.     STEP_EXECUTION_ID BIGINT  NOT NULL PRIMARY KEY ,
  48.     VERSION BIGINT NOT NULL,
  49.     STEP_NAME VARCHAR(100) NOT NULL,
  50.     JOB_EXECUTION_ID BIGINT NOT NULL,
  51.     START_TIME DATETIME NOT NULL ,
  52.     END_TIME DATETIME DEFAULT NULL ,
  53.     STATUS VARCHAR(10) ,
  54.     COMMIT_COUNT BIGINT ,
  55.     READ_COUNT BIGINT ,
  56.     FILTER_COUNT BIGINT ,
  57.     WRITE_COUNT BIGINT ,
  58.     READ_SKIP_COUNT BIGINT ,
  59.     WRITE_SKIP_COUNT BIGINT ,
  60.     PROCESS_SKIP_COUNT BIGINT ,
  61.     ROLLBACK_COUNT BIGINT ,
  62.     EXIT_CODE VARCHAR(2500) ,
  63.     EXIT_MESSAGE VARCHAR(2500) ,
  64.     LAST_UPDATED DATETIME,
  65.     constraint JOB_EXEC_STEP_FK foreign key (JOB_EXECUTION_ID)
  66.     references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
  67. ) ENGINE=InnoDB;
  68.  
  69. -- 该BATCH_STEP_EXECUTION_CONTEXT表包含ExecutionContext与Step相关的所有信息
  70. CREATE TABLE BATCH_STEP_EXECUTION_CONTEXT  (
  71.     STEP_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY,
  72.     SHORT_CONTEXT VARCHAR(2500) NOT NULL,
  73.     SERIALIZED_CONTEXT TEXT ,
  74.     constraint STEP_EXEC_CTX_FK foreign key (STEP_EXECUTION_ID)
  75.     references BATCH_STEP_EXECUTION(STEP_EXECUTION_ID)
  76. ) ENGINE=InnoDB;
  77.  
  78. -- 该表包含ExecutionContext与Job相关的所有信息
  79. CREATE TABLE BATCH_JOB_EXECUTION_CONTEXT  (
  80.     JOB_EXECUTION_ID BIGINT NOT NULL PRIMARY KEY,
  81.     SHORT_CONTEXT VARCHAR(2500) NOT NULL,
  82.     SERIALIZED_CONTEXT TEXT ,
  83.     constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
  84.     references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
  85. ) ENGINE=InnoDB;
  86.  
  87. CREATE TABLE BATCH_STEP_EXECUTION_SEQ (
  88.     ID BIGINT NOT NULL,
  89.     UNIQUE_KEY CHAR(1) NOT NULL,
  90.     constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
  91. ) ENGINE=InnoDB;
  92. INSERT INTO BATCH_STEP_EXECUTION_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_STEP_EXECUTION_SEQ);
  93.  
  94. CREATE TABLE BATCH_JOB_EXECUTION_SEQ (
  95.     ID BIGINT NOT NULL,
  96.     UNIQUE_KEY CHAR(1) NOT NULL,
  97.     constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
  98. ) ENGINE=InnoDB;
  99. INSERT INTO BATCH_JOB_EXECUTION_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_JOB_EXECUTION_SEQ);
  100.  
  101. CREATE TABLE BATCH_JOB_SEQ (
  102.     ID BIGINT NOT NULL,
  103.     UNIQUE_KEY CHAR(1) NOT NULL,
  104.     constraint UNIQUE_KEY_UN unique (UNIQUE_KEY)
  105. ) ENGINE=InnoDB;
  106. INSERT INTO BATCH_JOB_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp where not exists(select * from BATCH_JOB_SEQ);

1.4 Student建表SQL及表数据：

  1. CREATE TABLE `student` (
  2.   `id` int(11) NOT NULL AUTO_INCREMENT,
  3.   `name` varchar(20) NOT NULL,
  4.   `age` int(11) NOT NULL,
  5.   `sex` varchar(20) NOT NULL,
  6.   `address` varchar(100) NOT NULL,
  7.   `cid` int(11) NOT NULL,
  8.   PRIMARY KEY (`id`) USING BTREE
  9. ) ENGINE=InnoDB AUTO_INCREMENT=19 DEFAULT CHARSET=utf8;

Student表中数据：

二、创建项目

2.1 创建Spring Boot项目：

2.2 填写项目名、包名等信息：

2.3 勾选依赖项

勾选如下红框标注的依赖项：

2.4 完成创建

点击Finish完成项目的创建：

2.5 Pom.xml依赖

项目最终依赖：

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-batch</artifactId>
  5.     </dependency>
  6.     <dependency>
  7.         <groupId>org.springframework.boot</groupId>
  8.         <artifactId>spring-boot-starter-data-jpa</artifactId>
  9.     </dependency>
  10.     <dependency>
  11.         <groupId>org.springframework.boot</groupId>
  12.         <artifactId>spring-boot-starter-web</artifactId>
  13.     </dependency>
  14.  
  15.     <dependency>
  16.         <groupId>mysql</groupId>
  17.         <artifactId>mysql-connector-java</artifactId>
  18.         <scope>runtime</scope>
  19.     </dependency>
  20.     <dependency>
  21.         <groupId>org.projectlombok</groupId>
  22.         <artifactId>lombok</artifactId>
  23.         <optional>true</optional>
  24.     </dependency>
  25.     <dependency>
  26.         <groupId>org.springframework.boot</groupId>
  27.         <artifactId>spring-boot-starter-test</artifactId>
  28.         <scope>test</scope>
  29.     </dependency>
  30.     <dependency>
  31.         <groupId>org.springframework.batch</groupId>
  32.         <artifactId>spring-batch-test</artifactId>
  33.         <scope>test</scope>
  34.     </dependency>
  35. </dependencies>

2.6 SpringBoot配置文件:

  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.cj.jdbc.Driver
  4.     url: jdbc:mysql://127.0.0.1:3306/springbatch?serverTimezone=Asia/Shanghai&characterEncoding=UTF-8&autoReconnect=true
  5.     hikari:
  6.       password: password
  7.       username: root
  8.   jpa:
  9.     open-in-view: true
  10.     show-sql: true
  11.     hibernate:
  12.       ddl-auto: update
  13.     database: mysql
  14.   # 禁止项目启动时运行job
  15.   batch:
  16.     job:
  17.       enabled: false

三、基于Spring Batch的批处理Demo

3.1 Demo项目结构：

3.2 Student Entity：

  1. package org.zero.example.springbatchdemo.model;
  2.  
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6.  
  7. import javax.persistence.*;
  8.  
  9. /**
  10.  * student 表格的实体类
  11.  * 
  12.  * @author 01
  13.  * @date 2019-02-24
  14.  **/
  15. @Data
  16. @Entity
  17. @Table(name = "student")
  18. @NoArgsConstructor
  19. @AllArgsConstructor
  20. public class Student {
  21.     @Id
  22.     @GeneratedValue(strategy = GenerationType.IDENTITY)
  23.     private Integer id;
  24.  
  25.     private String name;
  26.  
  27.     private Integer age;
  28.  
  29.     private String sex;
  30.  
  31.     private String address;
  32.  
  33.     private Integer cid;
  34. }

3.3 配置Spring的线程池

批处理作业和定时任务需要多线程，因此配置Spring的线程池，代码如下：

  1. package org.zero.example.springbatchdemo.config;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  6.  
  7. /**
  8.  * 配置任务线程池执行器
  9.  *
  10.  * @author 01
  11.  * @date 2019-02-24
  12.  **/
  13. @Configuration
  14. public class ExecutorConfiguration {
  15.  
  16.     @Bean
  17.     public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
  18.         ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
  19.         threadPoolTaskExecutor.setCorePoolSize(50);
  20.         threadPoolTaskExecutor.setMaxPoolSize(200);
  21.         threadPoolTaskExecutor.setQueueCapacity(1000);
  22.         threadPoolTaskExecutor.setThreadNamePrefix("Data-Job");
  23.  
  24.         return threadPoolTaskExecutor;
  25.     }
  26. }

3.4 配置作业监听器

实现一个作业监听器，批处理作业在执行前后会调用监听器的方法：

  1. package org.zero.example.springbatchdemo.task.listener;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.batch.core.BatchStatus;
  5. import org.springframework.batch.core.JobExecution;
  6. import org.springframework.batch.core.JobExecutionListener;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  9. import org.springframework.stereotype.Component;
  10.  
  11. /**
  12.  * 一个简单的job监听器
  13.  *
  14.  * @author 01
  15.  * @date 2019-02-24
  16.  **/
  17. @Slf4j
  18. @Component
  19. public class JobListener implements JobExecutionListener {
  20.     private final ThreadPoolTaskExecutor threadPoolTaskExecutor;
  21.     private long startTime;
  22.  
  23.     @Autowired
  24.     public JobListener(ThreadPoolTaskExecutor threadPoolTaskExecutor) {
  25.         this.threadPoolTaskExecutor = threadPoolTaskExecutor;
  26.     }
  27.  
  28.     /**
  29.      * 该方法会在job开始前执行
  30.      */
  31.     @Override
  32.     public void beforeJob(JobExecution jobExecution) {
  33.         startTime = System.currentTimeMillis();
  34.         log.info("job before " + jobExecution.getJobParameters());
  35.     }
  36.  
  37.     /**
  38.      * 该方法会在job结束后执行
  39.      */
  40.     @Override
  41.     public void afterJob(JobExecution jobExecution) {
  42.         log.info("JOB STATUS : {}", jobExecution.getStatus());
  43.         if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
  44.             log.info("JOB FINISHED");
  45.             threadPoolTaskExecutor.destroy();
  46.         } else if (jobExecution.getStatus() == BatchStatus.FAILED) {
  47.             log.info("JOB FAILED");
  48.         }
  49.         log.info("Job Cost Time : {}/ms", (System.currentTimeMillis() - startTime));
  50.     }
  51. }

3.5 配置Job

配置一个最基本的Job，Job是真正进行批处理业务的地方。一个Job 通常由一个或多个Step组成(基本就像是一个工作流)；而一个Step通常由三部分组成（读入数据：ItemReader，处理数据：ItemProcessor，写入数据：ItemWriter）。代码如下： 
  1. package org.zero.example.springbatchdemo.task.job;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  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.core.launch.support.RunIdIncrementer;
  9. import org.springframework.batch.item.ItemProcessor;
  10. import org.springframework.batch.item.ItemReader;
  11. import org.springframework.batch.item.ItemWriter;
  12. import org.springframework.batch.item.database.JpaPagingItemReader;
  13. import org.springframework.batch.item.database.orm.JpaNativeQueryProvider;
  14. import org.springframework.beans.factory.annotation.Autowired;
  15. import org.springframework.stereotype.Component;
  16. import org.zero.example.springbatchdemo.model.Student;
  17. import org.zero.example.springbatchdemo.task.listener.JobListener;
  18.  
  19. import javax.persistence.EntityManagerFactory;
  20.  
  21. /**
  22.  * 配置一个最基本的Job
  23.  *
  24.  * @author 01
  25.  * @date 2019-02-24
  26.  **/
  27. @Slf4j
  28. @Component
  29. public class DataBatchJob {
  30.     /**
  31.      * Job构建工厂，用于构建Job
  32.      */
  33.     private final JobBuilderFactory jobBuilderFactory;
  34.  
  35.     /**
  36.      * Step构建工厂，用于构建Step
  37.      */
  38.     private final StepBuilderFactory stepBuilderFactory;
  39.  
  40.     /**
  41.      * 实体类管理工工厂，用于访问表格数据
  42.      */
  43.     private final EntityManagerFactory emf;
  44.  
  45.     /**
  46.      * 自定义的简单Job监听器
  47.      */
  48.     private final JobListener jobListener;
  49.  
  50.     @Autowired
  51.     public DataBatchJob(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
  52.                         EntityManagerFactory emf, JobListener jobListener) {
  53.         this.jobBuilderFactory = jobBuilderFactory;
  54.         this.stepBuilderFactory = stepBuilderFactory;
  55.         this.emf = emf;
  56.         this.jobListener = jobListener;
  57.     }
  58.  
  59.     /**
  60.      * 一个最基础的Job通常由一个或者多个Step组成
  61.      */
  62.     public Job dataHandleJob() {
  63.         return jobBuilderFactory.get("dataHandleJob").
  64.                 incrementer(new RunIdIncrementer()).
  65.                 // start是JOB执行的第一个step
  66.                         start(handleDataStep()).
  67.                 // 可以调用next方法设置其他的step，例如：
  68.                 // next(xxxStep()).
  69.                 // next(xxxStep()).
  70.                 // ...
  71.                 // 设置我们自定义的JobListener
  72.                         listener(jobListener).
  73.                         build();
  74.     }
  75.  
  76.     /**
  77.      * 一个简单基础的Step主要分为三个部分
  78.      * ItemReader : 用于读取数据
  79.      * ItemProcessor : 用于处理数据
  80.      * ItemWriter : 用于写数据
  81.      */
  82.     private Step handleDataStep() {
  83.         return stepBuilderFactory.get("getData").
  84.                 // <输入对象, 输出对象>  chunk通俗的讲类似于SQL的commit; 这里表示处理(processor)100条后写入(writer)一次
  85.                         <Student, Student>chunk(100).
  86.                 // 捕捉到异常就重试,重试100次还是异常,JOB就停止并标志失败
  87.                         faultTolerant().retryLimit(3).retry(Exception.class).skipLimit(100).skip(Exception.class).
  88.                 // 指定ItemReader对象
  89.                         reader(getDataReader()).
  90.                 // 指定ItemProcessor对象
  91.                         processor(getDataProcessor()).
  92.                 // 指定ItemWriter对象
  93.                         writer(getDataWriter()).
  94.                         build();
  95.     }
  96.  
  97.     /**
  98.      * 读取数据
  99.      *
  100.      * @return ItemReader Object
  101.      */
  102.     private ItemReader<? extends Student> getDataReader() {
  103.         // 读取数据,这里可以用JPA,JDBC,JMS 等方式读取数据
  104.         JpaPagingItemReader<Student> reader = new JpaPagingItemReader<>();
  105.  
  106.         try {
  107.             // 这里选择JPA方式读取数据
  108.             JpaNativeQueryProvider<Student> queryProvider = new JpaNativeQueryProvider<>();
  109.             // 一个简单的 native SQL
  110.             queryProvider.setSqlQuery("SELECT * FROM student");
  111.             // 设置实体类
  112.             queryProvider.setEntityClass(Student.class);
  113.             queryProvider.afterPropertiesSet();
  114.  
  115.             reader.setEntityManagerFactory(emf);
  116.             // 设置每页读取的记录数
  117.             reader.setPageSize(3);
  118.             // 设置数据提供者
  119.             reader.setQueryProvider(queryProvider);
  120.             reader.afterPropertiesSet();
  121.  
  122.             // 所有ItemReader和ItemWriter实现都会在ExecutionContext提交之前将其当前状态存储在其中,
  123.             // 如果不希望这样做,可以设置setSaveState(false)
  124.             reader.setSaveState(true);
  125.         } catch (Exception e) {
  126.             e.printStackTrace();
  127.         }
  128.  
  129.         return reader;
  130.     }
  131.  
  132.     /**
  133.      * 处理数据
  134.      *
  135.      * @return ItemProcessor Object
  136.      */
  137.     private ItemProcessor<Student, Student> getDataProcessor() {
  138.         return student -> {
  139.             // 模拟处理数据，这里处理就是打印一下
  140.             log.info("processor data : " + student.toString());
  141.  
  142.             return student;
  143.         };
  144.     }
  145.  
  146.     /**
  147.      * 写入数据
  148.      *
  149.      * @return ItemWriter Object
  150.      */
  151.     private ItemWriter<Student> getDataWriter() {
  152.         return list -> {
  153.             for (Student student : list) {
  154.                 // 模拟写数据，为了演示的简单就不写入数据库了
  155.                 log.info("write data : " + student);
  156.             }
  157.         };
  158.     }
  159. }

3.6 执行Job

通常运行Job的方式有两种:

  1. 把Job对象注入到Spring容器里，Spring Batch默认在项目启动完成后就会运行容器里配置好的Job，如果配置了多个Job也可以通过配置文件去指定。

  1. 采用定时任务去运行Job。通过调用的方式主动去运行Job，需要使用到JobLauncher中的run方法。具体代码如下：

  1. package org.zero.example.springbatchdemo.task;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.batch.core.*;
  5. import org.springframework.batch.core.launch.JobLauncher;
  6. import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
  7. import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
  8. import org.springframework.batch.core.repository.JobRestartException;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.scheduling.annotation.Scheduled;
  11. import org.springframework.stereotype.Component;
  12. import org.zero.example.springbatchdemo.task.job.DataBatchJob;
  13.  
  14. /**
  15.  * 简单的定时任务
  16.  *
  17.  * @author 01
  18.  * @date 2019-02-24
  19.  **/
  20. @Slf4j
  21. @Component
  22. public class TimeTask {
  23.  
  24.     private final JobLauncher jobLauncher;
  25.     private final DataBatchJob dataBatchJob;
  26.  
  27.     @Autowired
  28.     public TimeTask(JobLauncher jobLauncher, DataBatchJob dataBatchJob) {
  29.         this.jobLauncher = jobLauncher;
  30.         this.dataBatchJob = dataBatchJob;
  31.     }
  32.  
  33.     // 定时任务，每十秒执行一次
  34.     @Scheduled(cron = "0/10 * * * * ?")
  35.     public void runBatch() throws JobParametersInvalidException, JobExecutionAlreadyRunningException,
  36.             JobRestartException, JobInstanceAlreadyCompleteException {
  37.         log.info("定时任务执行了...");
  38.         // 在运行一个job的时候需要添加至少一个参数，这个参数最后会被写到batch_job_execution_params表中，
  39.         // 不添加这个参数的话，job不会运行，并且这个参数在表中中不能重复，若设置的参数已存在表中，则会抛出异常，
  40.         // 所以这里才使用时间戳作为参数
  41.         JobParameters jobParameters = new JobParametersBuilder()
  42.                 .addLong("timestamp", System.currentTimeMillis())
  43.                 .toJobParameters();
  44.  
  45.         // 获取job并运行
  46.         Job job = dataBatchJob.dataHandleJob();
  47.         JobExecution execution = jobLauncher.run(job, jobParameters);
  48.         log.info("定时任务结束. Exit Status : {}", execution.getStatus());
  49.     }
  50. }

3.7 配置Spring Boot启动类

最后，在Spring Boot的启动类上加上两个注解@EnableBatchProcessing和@EnableScheduling，以开启批处理及定时任务，否则批处理和定时任务都不会执行，代码如下：

  1. package org.zero.example.springbatchdemo;
  2.  
  3. import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
  4. import org.springframework.boot.SpringApplication;
  5. import org.springframework.boot.autoconfigure.SpringBootApplication;
  6. import org.springframework.scheduling.annotation.EnableScheduling;
  7.  
  8. /**
  9.  * :@EnableBatchProcessing 用于开启批处理作业的配置
  10.  * :@EnableScheduling 用于开启定时任务的配置
  11.  *
  12.  * @author 01
  13.  * @date 2019-02-24
  14.  */
  15. @EnableScheduling
  16. @EnableBatchProcessing
  17. @SpringBootApplication
  18. public class SpringBatchDemoApplication {
  19.     public static void main(String[] args) {
  20.         SpringApplication.run(SpringBatchDemoApplication.class, args);
  21.     }
  22. }

启动项目，等待十秒，控制台输出日志如下，证明我们的批处理程序正常执行了：

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

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

  

闽ICP备14008679号