Skip to main content
 首页 » 编程设计

Spring Batch 之 Hello World教程

2022年07月19日145lovecherry

Spring Batch 之 Hello World教程

本文我们基于spring boot和spring batch 实现一个简单hello world入门批处理程序。如果你刚刚接触spring batch，这篇教程会让你花最短时间理解spring batch框架。

Spring Batch 框架介绍

开始代码之前，我们先了解框架中的核心组件，见下图：
在这里插入图片描述

  • 批处理过程有Job组成，job是封装整个批处理过程的实体。
  • Job有一个或多个Step组成。大多数情况下，step负责读数据(ItemReader)，处理数据(ItemProcessor)以及写数据(ItemWriter)。
  • JobLauncher 负责启动job。
  • JobRepository负责存储关于配置和执行Job的元信息。

理解基本概念之后，我们通过一个简单Hello World程序来说明框架如何工作。示例从person.csv文件读取用户的firstName 和 lastName，然后给每个用户增加问候（hello），最后把结果写入 greetings.txt文件。

项目依赖

通过idea或Spring Initializr 新建spring boot 项目，增加相应依赖：

plugins { 
    id 'org.springframework.boot' version '2.1.3.RELEASE' 
    id 'java' 
} 
 
apply plugin: 'io.spring.dependency-management' 
 
group = 'com.dataz.batch' 
version = '0.0.1-SNAPSHOT' 
sourceCompatibility = '1.8' 
 
configurations { 
    compileOnly { 
        extendsFrom annotationProcessor 
    } 
} 
 
repositories { 
    mavenCentral() 
} 
 
dependencies { 
    implementation 'org.springframework.boot:spring-boot-starter-batch' 
    compileOnly 'org.projectlombok:lombok' 
    runtimeOnly 'org.springframework.boot:spring-boot-devtools' 
    annotationProcessor 'org.projectlombok:lombok' 
    testImplementation 'org.springframework.boot:spring-boot-starter-test' 
    testImplementation 'org.springframework.batch:spring-batch-test' 
}

spring-boot-starter-batch 引入Spring Boot 和 Spring Batch 依赖.

spring-boot-starter-test 引入Spring Boot 应用的测试依赖，包括JUnit, Hamcrest 和 Mockito.

spring-batch-test 引入测试Spring batch job和step的工具类.

Spring Boot 设置

我们使用了Spring Boot可以让Spring Batch应用立刻运行。
@SpringBootApplication注解包括@Configuration, @EnableAutoConfiguration, @ComponentScan 三个注解，消除或简化配置。

Spring Batch缺省使用数据库存储配置批处理job的元数据。为了简化，我们不使用数据库，而使用基于内存（Map）存储。

spring-boot-starter-batch 自动引入spring-boot-starter-jdbc,后者会尝试实例化datasource。在@SpringBootApplication注解上增加exclude = {DataSourceAutoConfiguration.class}，避免Spring Boot 自动配置Datasource至数据库连接。

完整 HelloWorldApplication 代码如下：

package com.dataz.batch.helloworld; 
 
import org.springframework.boot.SpringApplication; 
import org.springframework.boot.autoconfigure.SpringBootApplication; 
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; 
 
@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class}) 
public class HelloWorldApplication { 
 
    public static void main(String[] args) { 
        SpringApplication.run(HelloWorldApplication.class, args); 
    } 
}

创建域对象模型

处理数据之前，一般需要把数据映射为域对象。我们的数据存储在 src/resources/csv/person.csv，每行包括内容如下：

John, Doe 
Jane, Doe

为了映射数据至Person对象，下面定义Person类：

package com.dataz.batch.helloworld.dto; 
 
import lombok.Data; 
 
@Data 
public class Person { 
    private String firstName; 
    private String lastName; 
}

配置Spring Batch

定义BatchConfig类：

package com.dataz.batch.helloworld.configuration; 
 
import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer; 
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; 
import org.springframework.context.annotation.Configuration; 
 
import javax.sql.DataSource; 
 
@Configuration 
@EnableBatchProcessing 
public class BatchConfig extends DefaultBatchConfigurer { 
 
    @Override 
    public void setDataSource(DataSource dataSource) { 
        // initialize will use a Map based JobRepository (instead of database) 
    } 
}

我们创建BatchConfig 类用于配置 Spring Batch。@Configuration注解表明作为bead定义配置类。
@EnableBatchProcessing注解启用必要的Spring Batch特性，也提供构建job的基本配置。该注解功能强大，背后为我们做了很多事情，其创建下列bean：

  • JobRepository (bean名为 “jobRepository”)
  • JobLauncher (bean名为 “jobLauncher”)
  • JobRegistry (bean名为 “jobRegistry”)
  • JobExplorer (bean名为 “jobExplorer”)
  • PlatformTransactionManager (bean名为 “transactionManager”)
  • JobBuilderFactory (bean名为 “jobBuilders”) 用于构建job对象，避免给每个job注入jobRepository
  • StepBuilderFactory (bean名为 “stepBuilders”) 用于构建job对象，避免给每个step注入jobRepository 和事物管理bean

因为Spring Batch使用Map存储 ,需要继承DefaultBatchConfigurer，并重写setDataSource方法，不设置数据源。让自动配置使用基于Map的JobRepository.

配置批处理Job

下面继续配置批处理Job，我们创建HelloWorldJobConfig配置类并增加@Configuration注解：

package com.dataz.batch.helloworld.configuration; 
 
import com.dataz.batch.helloworld.dto.Person; 
import com.dataz.batch.helloworld.item.PersonItemProcessor; 
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.item.file.FlatFileItemReader; 
import org.springframework.batch.item.file.FlatFileItemWriter; 
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder; 
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder; 
import org.springframework.batch.item.file.transform.PassThroughLineAggregator; 
import org.springframework.context.annotation.Bean; 
import org.springframework.context.annotation.Configuration; 
import org.springframework.core.io.ClassPathResource; 
import org.springframework.core.io.FileSystemResource; 
 
@Configuration 
public class HelloWorldJobConfig { 
 
    @Bean 
    public Job helloWorlJob(JobBuilderFactory jobBuilders, 
                            StepBuilderFactory stepBuilders) { 
        return jobBuilders.get("helloWorldJob") 
                          .start(helloWorldStep(stepBuilders)).build(); 
    } 
 
    @Bean 
    public Step helloWorldStep(StepBuilderFactory stepBuilders) { 
        return stepBuilders.get("helloWorldStep") 
            .<Person, String>chunk(10).reader(reader()) 
                                      .processor(processor()).writer(writer()).build(); 
    } 
 
    @Bean 
    public FlatFileItemReader<Person> reader() { 
        return new FlatFileItemReaderBuilder<Person>() 
            .name("personItemReader") 
            .resource(new ClassPathResource("csv/person.csv")) 
            .delimited().names(new String[] {"firstName", "lastName"}) 
            .targetType(Person.class).build(); 
    } 
 
    @Bean 
    public PersonItemProcessor processor() { 
        return new PersonItemProcessor(); 
    } 
 
    @Bean 
    public FlatFileItemWriter<String> writer() { 
        return new FlatFileItemWriterBuilder<String>() 
            .name("greetingItemWriter") 
            .resource(new FileSystemResource( 
                "target/test-outputs/greetings.txt")) 
            .lineAggregator(new PassThroughLineAggregator<>()).build(); 
    } 
}

helloWorlJob bean 使用JobBuilderFactory 通过传递job和step的名称创建job，运行时通过名称进行查找。
需要注意的是jobBuilders 和 stepBuilders bean 会被自动注入。它们在上一节中通过@EnableBatchProcessing负责创建。

helloWorldStep bean 定义step中执行的不同item，这里使用StepBuilderFactory创建step。

首先传入step名称，然后使用chunk()方法指定每个事物处理数据项数量，同时也指明输入类型为person，输出为string。最后给step增加reader、process以及writer。

我们使用FlatFileItemReader读perons.csv文件，该类提供了基本的文件文件处理功能。通过FlatFileItemReaderBuilder创建FlatFileItemReader实现，首先传入reader名称，接着传入处理资源，最后是输出类型Person。FlatFileItemReader处理文件需要额外的信息，分隔符（默认为逗号）及字段映射信息（names方法指定），表明第一列映射到Person的firstName，第二列映射到lastName。

PersonItemProcessor负责处理数据，其转换每个Person对象为一个字符串。
数据处理完通过FlatFileItemWriter 组件写入一个文本文件。FlatFileItemWriterBuilder 创建FlatFileItemWriter组件，指定名称和对应处理资源（greeting.txt文件）。FlatFileItemWriter需要知道如何生成单个字符串，因为本例中输出以及是一个字符串，可以使用PassThroughLineAggregator，这是最基本的实现，其假设对象已经是一个字符串。

处理数据

大多数情况下，job执行过程中需要进行数据处理。我们示例中把Person转成字符串。PersonItemProcessor类实现ItemProcessor接口，在process方法中实现具体业务。

package com.dataz.batch.helloworld.item; 
 
import com.dataz.batch.helloworld.dto.Person; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.batch.item.ItemProcessor; 
 
public class PersonItemProcessor 
    implements ItemProcessor<Person, String> { 
 
    private static final Logger LOGGER = 
        LoggerFactory.getLogger(PersonItemProcessor.class); 
 
    @Override 
    public String process(Person person) { 
        String greeting = "Hello " + person.getFirstName() + " " 
            + person.getLastName() + "!"; 
 
        LOGGER.info("converting '{}' into '{}'", person, greeting); 
        return greeting; 
    } 
}

运行或测试Batch示例

我们可以直接运行HelloWorldApplication类。也可以通过单元测试运行，首先定义单元测试配置类：

package com.dataz.batch.helloworld.configuration; 
 
import org.springframework.batch.core.Job; 
import org.springframework.batch.test.JobLauncherTestUtils; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.context.annotation.Bean; 
import org.springframework.context.annotation.Configuration; 
import org.springframework.context.annotation.Import; 
 
@Configuration 
@Import({BatchConfig.class, HelloWorldJobConfig.class}) 
public class BatchTestConfig { 
 
    @Autowired 
    private Job helloWorlJob; 
 
    @Bean 
    JobLauncherTestUtils jobLauncherTestUtils() { 
        JobLauncherTestUtils jobLauncherTestUtils = 
            new JobLauncherTestUtils(); 
        jobLauncherTestUtils.setJob(helloWorlJob); 
 
        return jobLauncherTestUtils; 
    } 
}

这里主要定义JobLauncherTestUtils bean，用于单元测试。

下面定义测试类：

package com.dataz.batch.helloworld; 
 
import com.dataz.batch.helloworld.configuration.BatchTestConfig; 
import org.junit.Test; 
import org.junit.runner.RunWith; 
import org.springframework.batch.core.JobExecution; 
import org.springframework.batch.test.JobLauncherTestUtils; 
import org.springframework.beans.factory.annotation.Autowired; 
import org.springframework.boot.test.context.SpringBootTest; 
import org.springframework.test.context.junit4.SpringRunner; 
 
import static org.assertj.core.api.Assertions.assertThat; 
 
@RunWith(SpringRunner.class) 
@SpringBootTest(classes = {BatchTestConfig.class}) 
public class HelloWorldApplicationTests { 
 
    @Autowired 
    private JobLauncherTestUtils jobLauncherTestUtils; 
 
    @Test 
    public void testHelloWorldJob() throws Exception { 
        JobExecution jobExecution = jobLauncherTestUtils.launchJob(); 
        assertThat(jobExecution.getExitStatus().getExitCode()).isEqualTo("COMPLETED"); 
    } 
}

注意这里使用@RunWith(SpringRunner.class)和@SpringBootTest(classes = {BatchTestConfig.class})注解，通知junit使用Spring test和spring Boot支持。运行结果如下：

2019-03-29 16:16:23.210  INFO 9004 --- [main] c.d.b.h.HelloWorldApplicationTests       : Started HelloWorldApplicationTests in 7.919 seconds (JVM running for 16.449) 
2019-03-29 16:16:24.708  INFO 9004 --- [main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=helloWorldJob]] launched with the following parameters: [{random=360857}] 
2019-03-29 16:16:24.724  INFO 9004 --- [main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [helloWorldStep] 
2019-03-29 16:16:24.919  INFO 9004 --- [main] c.d.b.h.item.PersonItemProcessor         : converting 'Person(firstName=John, lastName=Doe)' into 'Hello John Doe!' 
2019-03-29 16:16:24.921  INFO 9004 --- [main] c.d.b.h.item.PersonItemProcessor         : converting 'Person(firstName=Jane, lastName=Doe)' into 'Hello Jane Doe!' 
2019-03-29 16:16:24.937  INFO 9004 --- [main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=helloWorldJob]] completed with the following parameters: [{random=360857}] and the following status: [COMPLETED]

最终你能看到结果在文件在target/test-outputs/greetings.txt，其结果如下：

Hello John Doe! 
Hello Jane Doe!

总结

本文通过简单示例说明了Spring Batch的原理及关键概念及组件。包括job、step以及处理环节（ItemReader、ItemProcessor、ItemWriter)，以及初始化Spring Batch运行环境和单元测试配置。


本文参考链接：https://blog.csdn.net/neweastsun/article/details/88953306
  文章来源：
  https://www.itguest.com/post/agba2b1.html
阅读延展