java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > Spring Batch批处理

Spring Batch轻量级批处理框架实战

作者:Mr.Ymx

本文主要介绍了Spring Batch轻量级批处理框架实战,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下

1 实战前的理论基础

1.1 Spring Batch是什么

Spring Batch 是一个轻量级、全面的批处理框架,旨在支持开发对企业系统日常运营至关重要的强大的批处理应用程序。同时使开发人员在必要时可以轻松访问和利用更先进的企业服务。Spring Batch 不是调度框架,它旨在与调度程序一起工作,而不是取代调度程序。

1.2 Spring Batch能做什么

业务场景:

总之Spring batch可以做的:

1.3 基础架构

在这里插入图片描述

1.4 核心概念和抽象

在这里插入图片描述

核心概念:一个 Job 有一对多的Step,每个步骤都正好有一个 ItemReader、一个ItemProcessor和 一个ItemWriter。需要启动作业(使用 JobLauncher),并且需要存储有关当前运行进程的元数据(在 中 JobRepository)。

2 各个组件介绍

2.1 Job

Job是封装了整个批处理过程的实体。与其他 Spring 项目一样,一个Job与 XML 配置文件或基于 Java 的配置连接在一起。这种配置可以被称为“作业配置”。

在这里插入图片描述

可配置项:

2.2 Step

一个Step是一个域对象,它封装了批处理作业的一个独立的、连续的阶段。因此,每个 Job 完全由一个或多个步骤组成。一个Step包含定义和控制实际批处理所需的所有信息。

在这里插入图片描述

一个StepExecution代表一次尝试执行一个StepStepExecution 每次Step运行时都会创建一个新的,类似于JobExecution

2.3 ExecutionContext

一个ExecutionContext表示由框架持久化和控制的键/值对的集合,以允许开发人员有一个地方来存储范围为StepExecution对象或JobExecution对象的持久状态。

2.4 JobRepository

JobRepository是上述所有 Stereotypes 的持久性机制。它提供了CRUD操作JobLauncherJob以及Step实现。当 Job第一次启动,一个JobExecution被从库中获得,并且,执行的过程中,StepExecutionJobExecution实施方式是通过将它们传递到存储库持续。

使用 Java 配置时,@EnableBatchProcessing注解提供了一个 JobRepository作为开箱即用自动配置的组件之一。

2.5 JobLauncher

JobLauncher表示一个简单的接口,用于Job使用给定的 集合 启动JobParameters,如以下示例所示:

public interface JobLauncher {

public JobExecution run(Job job, JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException,
                   JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

期望实现JobExecution从 中 获得有效JobRepository并执行Job

2.6 Item Reader

ItemReader是一种抽象,表示一次检索Step一个项目的输入。当ItemReader用完它可以提供的项目时,它通过返回来表明这一点null

2.7 Item Writer

ItemWriter是一种抽象,表示一次一个Step、一批或一大块项目的输出。通常, anItemWriter不知道它接下来应该接收的输入,并且只知道在其当前调用中传递的项目。

2.8 Item Processor

ItemProcessor是表示项目的业务处理的抽象。当ItemReader读取一个项目并ItemWriter写入它们时,它 ItemProcessor提供了一个访问点来转换或应用其他业务处理。如果在处理该项目时确定该项目无效,则返回 null表示不应写出该项目。

3 Spring Batch实战

下面就利用我们所学的理论实现一个最简单的Spring Batch批处理项目

3.1 依赖和项目结构以及配置文件

依赖

<!--Spring batch-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<!-- web依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- lombok-->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.20</version>
</dependency>
<!--  mysql-->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>
<!--  mybatis-->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.2.0</version>
</dependency>

项目结构

在这里插入图片描述

配置文件

server.port=9000
spring.datasource.url=jdbc:mysql://localhost:3306/test
spring.datasource.username=root
spring.datasource.password=12345
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

3.2 代码和数据表

数据表

CREATE TABLE `student` (
    `id` int(100) NOT NULL AUTO_INCREMENT,
    `name` varchar(45) DEFAULT NULL,
    `age` int(2) DEFAULT NULL,
    `address` varchar(45) DEFAULT NULL,
    PRIMARY KEY (`id`),
    UNIQUE KEY `id_UNIQUE` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=203579 DEFAULT CHARSET=utf8 ROW_FORMAT=REDUNDANT

Student实体类

/**
 * @desc: Student实体类
 * @author: YanMingXin
 * @create: 2021/10/15-12:17
 **/
@Data
@Accessors(chain = true)
@NoArgsConstructor
@AllArgsConstructor
@ToString
@TableName("student")
public class Student {

    @TableId(value = "id", type = IdType.AUTO)
    private Long sId;

    @TableField("name")
    private String sName;

    @TableField("age")
    private Integer sAge;

    @TableField("address")
    private String sAddress;

}

Mapper层

/**
 * @desc: Mapper层
 * @author: YanMingXin
 * @create: 2021/10/15-12:17
 **/
@Mapper
@Repository
public interface StudentDao extends BaseMapper<Student> {
}

模拟数据库(文件)中读取类

/**
 * @desc: 模拟数据库中读取
 * @author: YanMingXin
 * @create: 2021/10/16-10:13
 **/
public class StudentVirtualDao {

    /**
     * 模拟从数据库中读取
     *
     * @return
     */
    public List<Student> getStudents() {
        ArrayList<Student> students = new ArrayList<>();
        students.add(new Student(1L, "zs", 23, "Beijing"));
        students.add(new Student(2L, "ls", 23, "Beijing"));
        students.add(new Student(3L, "ww", 23, "Beijing"));
        students.add(new Student(4L, "zl", 23, "Beijing"));
        students.add(new Student(5L, "mq", 23, "Beijing"));
        students.add(new Student(6L, "gb", 23, "Beijing"));
        students.add(new Student(7L, "lj", 23, "Beijing"));
        students.add(new Student(8L, "ss", 23, "Beijing"));
        students.add(new Student(9L, "zsdd", 23, "Beijing"));
        students.add(new Student(10L, "zss", 23, "Beijing"));
        return students;
    }
}

Service层接口

/**
 * @desc:
 * @author: YanMingXin
 * @create: 2021/10/15-12:16
 **/
public interface StudentService {

    List<Student> selectStudentsFromDB();

    void insertStudent(Student student);
}

Service层实现类

/**
 * @desc: Service层实现类
 * @author: YanMingXin
 * @create: 2021/10/15-12:16
 **/
@Service
public class StudentServiceImpl implements StudentService {

    @Autowired
    private StudentDao studentDao;

    @Override
    public List<Student> selectStudentsFromDB() {
        return studentDao.selectList(null);
    }

    @Override
    public void insertStudent(Student student) {
        studentDao.insert(student);
    }
}

最核心的配置类BatchConfiguration

/**
 * @desc: BatchConfiguration
 * @author: YanMingXin
 * @create: 2021/10/15-12:25
 **/
@Configuration
@EnableBatchProcessing
@SuppressWarnings("all")
public class BatchConfiguration {

    /**
     * 注入JobBuilderFactory
     */
    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    /**
     * 注入StepBuilderFactory
     */
    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    /**
     * 注入JobRepository
     */
    @Autowired
    public JobRepository jobRepository;

    /**
     * 注入JobLauncher
     */
    @Autowired
    private JobLauncher jobLauncher;

    /**
     * 注入自定义StudentService
     */
    @Autowired
    private StudentService studentService;

    /**
     * 注入自定义job
     */
    @Autowired
    private Job studentJob;

    /**
     * 封装writer bean
     *
     * @return
     */
    @Bean
    public ItemWriter<Student> writer() {
        ItemWriter<Student> writer = new ItemWriter() {
            @Override
            public void write(List list) throws Exception {
                //debug发现是嵌套的List reader的线程List嵌套真正的List
                list.forEach((stu) -> {
                    for (Student student : (ArrayList<Student>) stu) {
                        studentService.insertStudent(student);
                    }
                });
            }
        };
        return writer;
    }

    /**
     * 封装reader bean
     *
     * @return
     */
    @Bean
    public ItemReader<Student> reader() {
        ItemReader<Student> reader = new ItemReader() {
            @Override
            public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                //模拟数据获取
                StudentVirtualDao virtualDao = new StudentVirtualDao();
                return virtualDao.getStudents();
            }
        };
        return reader;
    }

    /**
     * 封装processor bean
     *
     * @return
     */
    @Bean
    public ItemProcessor processor() {
        ItemProcessor processor = new ItemProcessor() {
            @Override
            public Object process(Object o) throws Exception {
                //debug发现o就是reader单次单线程读取的数据
                return o;
            }
        };
        return processor;
    }

    /**
     * 封装自定义step
     *
     * @return
     */
    @Bean
    public Step studentStepOne() {
        return stepBuilderFactory.get("studentStepOne")
            .chunk(1)
            .reader(reader()) //加入reader
            .processor(processor())  //加入processor
            .writer(writer())//加入writer
            .build();
    }

    /**
     * 封装自定义job
     *
     * @return
     */
    @Bean
    public Job studentJob() {
        return jobBuilderFactory.get("studentJob")
            .flow(studentStepOne())//加入step
            .end()
            .build();
    }


    /**
     * 使用spring 定时任务执行
     */
    @Scheduled(fixedRate = 5000)
    public void printMessage() {
        try {
            JobParameters jobParameters = new JobParametersBuilder()
                .addLong("time", System.currentTimeMillis())
                .toJobParameters();
            jobLauncher.run(studentJob, jobParameters);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.3 测试

在这里插入图片描述

项目启动1s之后

在这里插入图片描述

看数据库,除了我们实体类定义的表以外多出来这么多表,这些表都是spring batch自带的记录日志和错误的表,具体的字段含义的有待研究

在这里插入图片描述

4 实战后的总结

Spring Batch有非常快的写入和读取速度,但是带来的影响就是非常耗费内存和数据库连接池的资源如果使用不好的话还会发生异常,因此我们要进行正确的配置,接下来我们进行简单的源码探究:

4.1 JobBuilderFactory

job的获取使用了简单工厂模式和建造者模式JobBuilderFactory获取JobBuilder在经过配置返回一个job对象的实例,该实例就是Spring Batch中最顶级的组件,包含了n和step

public class JobBuilderFactory {

   private JobRepository jobRepository;

   public JobBuilderFactory(JobRepository jobRepository) {
      this.jobRepository = jobRepository;
   }
   //返回JobBuilder
   public JobBuilder get(String name) {
      JobBuilder builder = new JobBuilder(name).repository(jobRepository);
      return builder;
   }
}

jobBuilder类

public class JobBuilder extends JobBuilderHelper<JobBuilder> {

   /**
    * 为指定名称的作业创建一个新的构建器
    */
   public JobBuilder(String name) {
      super(name);
   }

   /**
    * 创建将执行步骤或步骤序列的新作业构建器。
    */
   public SimpleJobBuilder start(Step step) {
      return new SimpleJobBuilder(this).start(step);
   }

   /**
    * 创建将执行流的新作业构建器。
    */
   public JobFlowBuilder start(Flow flow) {
      return new FlowJobBuilder(this).start(flow);
   }

   /**
    * 创建将执行步骤或步骤序列的新作业构建器
    */
   public JobFlowBuilder flow(Step step) {
      return new FlowJobBuilder(this).start(step);
   }
}

4.2 StepBuilderFactory

直接看StepBuilder类

public class StepBuilder extends StepBuilderHelper<StepBuilder> {

   public StepBuilder(String name) {
      super(name);
   }

   /**
    * 用自定义微线程构建步骤,不一定是项处理。  
    */
   public TaskletStepBuilder tasklet(Tasklet tasklet) {
      return new TaskletStepBuilder(this).tasklet(tasklet);
   }

   /**
    * 构建一个步骤,按照提供的大小以块的形式处理项。为了将这一步扩展到容错,
    * 在构建器上调用SimpleStepBuilder的 faultolerant()方法。
    * @param <I> 输入类型
    * @param <O> 输出类型
    */
   public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
      return new SimpleStepBuilder<I, O>(this).chunk(chunkSize);
   }

   public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
      return new SimpleStepBuilder<I, O>(this).chunk(completionPolicy);
   }

   public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) {
      return new PartitionStepBuilder(this).partitioner(stepName, partitioner);
   }

   public PartitionStepBuilder partitioner(Step step) {
      return new PartitionStepBuilder(this).step(step);
   }

   public JobStepBuilder job(Job job) {
      return new JobStepBuilder(this).job(job);
   }

   /**
    * 创建将执行流的新步骤构建器。
    */
   public FlowStepBuilder flow(Flow flow) {
      return new FlowStepBuilder(this).flow(flow);
   }
}

参考文档:

https://docs.spring.io/spring-batch/docs/4.3.x/reference/html/index.html

https://www.jdon.com/springbatch.html

到此这篇关于Spring Batch轻量级批处理框架实战的文章就介绍到这了,更多相关Spring Batch批处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文