tags： springbatch

[TOC]

1.引言

前面《数据批处理神器-Spring Batch(1)简介及使用场景》已经介绍了Spring Batch是一个轻量级，完善的批处理框架，它使用起来简单，方便，比较适合有点编程基础（特别是使用Spring及SpringBoot框架）的开发人员，针对业务编程，只需要关心具体的业务实现即可，把流程以及流程的控制交给Spring Batch就好。常言道”talk is cheap, show me the code“，下面我们就通过一个简单的hello world，进入Spring Batch的世界，通过这个示例，可以快速了解开发批处理的流程和Spring Batch开发用到的组件，为后续的操作打下基础。

2.开发环境

  • JDK: jdk1.8
  • Spring Boot: 2.1.4.RELEASE
  • Spring Batch:4.1.2.RELEASE
  • 开发IDE: IDEA
  • 构建工具Maven: 3.3.9
  • 日志组件logback:1.2.3
  • lombok:1.18.6

    3.helloworld开发

    3.1 helloworld说明

    本helloworld实现一个非常简单的功能，就是从数据组中读取字符串，把字符串转为大写，然后输出到控制台。如图：

快速了解组件-spring batch(2)之helloworld - 图1

字符串读写整个过程就是一个批重任务（Job），它只有一个步骤（Job Step），步骤里分为三个阶段，读数据（ItemReader）、处理数据（ItemProcessor）、写数据（ItemWriter）。

3.2 开发流程

开发的主要代码如下：

快速了解组件-spring batch(2)之helloworld - 图2

总体来说就是，通过Reader，Processor、Writer完成任务，结束后通过Listener进行监听，整个任务通过配置(BatchConfig)进行配置。

3.2.1 创建Spring Boot工程

直接使用Idea生成或在使用Spring Initializr生成即可，此处不详细说明。也可以直接使用我的代码示例。当前使用的Spring Boot版本是2.1.4.RELEASE

3.2.2 添加相关依赖

  • Spring Batch依赖 在使用spring-boot-starter-parent的情况下，直接添加以下依赖即可：

    1. <!-- 批处理框架-->
    2. <dependency>
    3.   <groupId>org.springframework.boot</groupId>
    4.   <artifactId>spring-boot-starter-batch</artifactId>
    5. </dependency>

    引用后，会引用两个jar包，一个是spring-batch-infrastructure，一个是spring-batch-core，版本是4.1.2.RELEASE。分别对应的是基础框架层和核心层。

  • 添加内存数据库H2 Spring Batch是需要数据库来存储任务的基本信息以及运行状态的，本例中不需要操作数据库逻辑，直接使用内存数据库H2即可。添加以下依赖：

    1. <dependency>
    2.   <groupId>com.h2database</groupId>
    3.   <artifactId>h2</artifactId>
    4. </dependency>

  • 添加测试及工具类依赖 为了简化开发，使用lombok进行处理。使用Spring Boot进行单元测试，添加依赖如下：

    1. <!-- 工具包：lombok -->
    2. <dependency>
    3.   <groupId>org.projectlombok</groupId>
    4.   <artifactId>lombok</artifactId>
    5.   <version>1.18.6</version>
    6. </dependency>
    7. <!-- 测试框架 -->
    8. <dependency>
    9.   <groupId>org.springframework.boot</groupId>
    10.   <artifactId>spring-boot-starter-test</artifactId>
    11.   <scope>test</scope>
    12. </dependency>

    3.2.3 开发读数据组件ItemReader

    添加完依赖后，就可以进入业务逻辑编程了。按Spring Batch的批处理流程，读数据ItemReader是第一步，当前示例中，我们的任务是从数组中读取数据。ItemReader是一个接口，开发人员直接实现此接口即可。此接口定义了核心方法read()，负责从给定的资源中读取可用的数据。具体实现如下：

    1. @Slf4j
    2. public class StringReader implements ItemReader<String> {
    3.   private String[] messages = {"aaa1","aaa2","aaa3","aaa4"};
    4.   private int count = 0;
    5.   @Override
    6.   public String read() throws UnexpectedInputException, ParseException, NonTransientResourceException {
    7.       if(count < messages.length){
    8.           String message = messages[count++];
    9.           log.debug(LogConstants.LOG_TAG + "read data:"+message);
    10.           return message;
    11.       }else{
    12.           log.debug(LogConstants.LOG_TAG + "read data end.");
    13.           count = 0;
    14.       }
    15.       return null;
    16.   }
    17. }

    说明：

  • （1）StringReader实现ItemReader接口；

  • （2）messages是数据源；
  • （3）count表示读取数据的下标，每读一次，下标自增，读取完后返回null表示结束。同时把count置为0，以方便下次读取。

  • （4）日志输出使用的是logback，结合lombok的@Slf4j注解，直接可使用log进行输出，简化操作。

    3.2.4 开发处理数据组件ItemProcessor

    读取数据后，返回的数据会流到ItemProcessor进行处理。同样，ItemProcessor是一个接口，要实现自己的处理逻辑，实现此接口即可。当然，如果没有ItemProcessor，读到的数据直接就到ItemWriter流程也是可以的。此处，Spring Batch有一个Chunk的概念，用于多次读，直到chunk指定的数量后，再统一给到processor和writer，以提高效率。本示例对于ItemProcessor的实现很简单，即把字符串转为大写。如下：

    1. @Slf4j
    2. public class ConvertProcessor implements ItemProcessor<String,String> {
    3.   @Autowired
    4.   private ConsoleService consoleService;
    5.   @Override
    6.   public String process(String data) {
    7.       String dataProcessed = consoleService.convert2UpperCase(data);
    8.       log.debug(LogConstants.LOG_TAG + data +" process data --> " + dataProcessed);
    9.       return dataProcessed;
    10.   }
    11. }

    说明：

  • 实现ItemProcessor接口，它有两个泛型，分别是I和O，I是读阶段获取的数据，O是提交给写阶段的数据。

  • 使用ConsoleService服务，对数据进行大写转换，里面的实现直接使用字符串的toUpperCase()方法

    3.2.5 开发写数据组件ItemWriter

    数据处理完后，会统一交给写组件(ItemWriter)进行写入。ItemWriter也是一个接口，核心方法是write方法，参数是数组。要实现自己的逻辑，实现此接口即可。本示例中，直接把数据输出到日志中即可。如下：
    1. @Slf4j
    2. public class ConsoleWriter implements ItemWriter<String> {
    3.   @Override
    4.   public void write(List<? extends String> list) {
    5.       for (String msg :list) {
    6.           log.debug(LogConstants.LOG_TAG + "write data: "+msg);
    7.       }
    8.   }
    9. }

    3.2.6 开发任务完成后的监听器JobExecutionListener

    数据写入到目标后，任务即结束，但有时候我们还需要在任务结束时去做一些其它工作，如清理数据，更新时间等，则需要在任务完成后进行逻辑处理。Spring Batch对于任务或步骤开始和结束都会提供监听，以便于开发人员实现监听逻辑。如通过继承JobExecutionListenerSupport，可以实现beforeJob和afterJob的监听，以实现开始任务前和结束任务后的处理。当前示例中，仅输出任务完成的日志。如下：
    1. @Slf4j
    2. public class ConsoleJobEndListener extends JobExecutionListenerSupport {
    3.   @Override
    4.   public void afterJob(JobExecution jobExecution) {
    5.       if(jobExecution.getStatus() == BatchStatus.COMPLETED){
    6.           log.debug("console batch job complete!");
    7.       }
    8.   }
    9. }

    3.2.7 配置完整任务

    经过上面的读、处理、写、任务完成后监听的操作，现在需要把它们组装在一起，形成一个完成的任务，使用Spring Boot，简单的使用几个配置即可完成任务的组装。任务及其相关组件的关系如下：
    快速了解组件-spring batch(2)之helloworld - 图3

创建配置文件ConsoleBatchConfig.java，具体代码如下：

  1. @Configuration
  2. @EnableBatchProcessing
  3. public class ConsoleBatchConfig {
  4.     @Autowired
  5.     public JobBuilderFactory jobBuilderFactory;
  6.     @Autowired
  7.     public StepBuilderFactory stepBuilderFactory;
  8.     @Bean
  9.     public Job consoleJob(Step consoleStep,JobExecutionListener consoleListener){
  10.         String funcName = Thread.currentThread().getStackTrace()[1].getMethodName();
  11.         return jobBuilderFactory.get(funcName).listener(consoleListener).flow(consoleStep)
  12.                 .end().build();
  13.     }
  14.     @Bean
  15.     public Step consoleStep(ItemReader stringReader,ItemProcessor convertProcessor
  16.             ,ItemWriter consoleWriter, CommonStepListener commonStepListener){
  17.         String funcName = Thread.currentThread().getStackTrace()[1].getMethodName();
  18.         return stepBuilderFactory.get(funcName).listener(commonStepListener)
  19.                 .<String,String>chunk(3).reader(stringReader).processor(convertProcessor)
  20.                 .writer(consoleWriter).build();
  21.     }
  22.     @Bean
  23.     public ItemReader stringReader(){return new StringReader();}
  24.     @Bean
  25.     public ItemWriter consoleWriter(){return new ConsoleWriter();}
  26.     @Bean
  27.     public ItemProcessor convertProcessor(){return new ConvertProcessor();}
  28.     @Bean
  29.     public JobExecutionListener consoleListener(){return new ConsoleJobEndListener();}
  30. }

说明：

  • 添加注解@Configuration及和@EnableBatchProcessing，标识为配置及启用Spring Batch的配置（可以直接使用JobBuilderFactory及StepBuilderFactory分别用于创建Job和Step）。
  • 创建ItemReader、ItemWriter、ItemProcessor、Listener对应的Bean，以供Step及Job的注入。
  • 使用stepBuilderFactory创建作业Step，其中chunk进行面向块的处理，即多次读取后再写入，提高效率。当前配置是3个为一个chunk。
  • 使用jobBuilderFactory添加step，创建任务。

  • 注意step和Job都需要有对应的名称(get方法确定)，此处直接使用方法名作为Job和Step的名称。

    3.2.8 测试批处理

    经过上面的步骤，已经完成Job的开发，测试则可使用两种方式，一个是编写Controller，以接口调用的方式运行job，一种编写单元测试。

  • Job的运行 通过JobLauncher的run方法来运行任务，run方法参数分别是Job和jobParameters，即已配置的Job及job运行的参数。每个任务的区分是通过任务名（jobName）和任务参数（jobParameters）作为区别的，即如果jobName和jobParameters相同，Spring Batch会认为是同一任务，若任务已运行成功，同一任务不会再运行。因此，一般来说，不同的任务，我们的jobParameters可以直接以时间作为参数，以便于区别。生成jobParameters。代码如下：

    1. JobParameters jobParameters = new JobParametersBuilder()
    2.               .addLong("time",System.currentTimeMillis())
    3.               .toJobParameters();

  • 编写单元测试 编写ConsoleJobTest，加载job，运行测试，如下所示：

    1. @RunWith(SpringRunner.class)
    2. @SpringBootTest(classes = {MainBootApplication.class,ConsoleBatchConfig.class})
    3. @Slf4j
    4. public class ConsoleJobTest {
    5.   @Autowired
    6.   private JobLauncher jobLauncher;
    7.   @Autowired
    8.   private Job consoleJob;
    9.   public void testConsoleJob2() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
    10.       //构建参数
    11.       JobParameters jobParameters = new JobParametersBuilder()
    12.               .addLong("time",System.currentTimeMillis())
    13.               .toJobParameters();
    14.       //执行任务
    15.       JobExecution run = jobLauncher.run(consoleJob, jobParameters);
    16.       ExitStatus exitStatus = run.getExitStatus();
    17.       log.debug(exitStatus.toString());
    18.   }
    19. }

    说明：引入SpringBootTest注解时，需要把Spring Batch任务也引入进来。

  • 执行结果输出 执行结果如下图所示：

快速了解组件-spring batch(2)之helloworld - 图4

从输出可知，由于设置的chunk是3，读取3个数据后，就统一给ItemProcessor进行大写转换处理，然后统一交给ItemWriter进行写入。执行完成后，Job的exitCode表示任务执行的状态，如果正常则为COMPLETED，失败则为FAILED。

4.总结

经过以上的操作步骤，即可完成批处理操作。关于任务的状态，流程的步骤（读、处理、写）均交给Spring Batch来完成，开发人员所做的工作是根据自己的业务逻辑编写具体的读数据、处理数据和写数据即可。希望通过本文，大家可以对Spring Batch的组件有清晰的了解。