史上最轻松入门之Spring Batch - 轻量级批处理框架实践

安宇雨 - 随手采集
2019-10-29 19:53:51
随手采集
0000-未整理-等待研究

点击上方☝SpringForAll社区，轻松关注！

及时获取有趣有料的技术文章

从 MariaDB 一张表内读 10 万条记录，经处理后写到 MongoDB 。

Batch 任务模型

具体实现

1、新建 Spring Boot 应用，依赖如下：

        <!-- Web 应用 -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>            <exclusions>                <exclusion>                    <groupId>org.springframework.boot</groupId>                    <artifactId>spring-boot-starter-logging</artifactId>                </exclusion>                <exclusion>                    <groupId>org.springframework.boot</groupId>                    <artifactId>spring-boot-starter-tomcat</artifactId>                </exclusion>            </exclusions>        </dependency>        <!-- Web 容器 undertow -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-undertow</artifactId>        </dependency>        <!-- 日志 Log4j2 -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-log4j2</artifactId>        </dependency>        <!-- MongoDB -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-data-mongodb</artifactId>        </dependency>        <!-- Brantch -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-batch</artifactId>        </dependency>        <!-- Mariadb 驱动 -->        <dependency>            <groupId>org.mariadb.jdbc</groupId>            <artifactId>mariadb-java-client</artifactId>            <version>2.0.2</version>        </dependency>        <!-- Lombok 代码简化 -->        <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <version>1.16.14</version>        </dependency>

2、创建一张表，并生成 10 万条数据

DROP TABLE people IF EXISTS;CREATE TABLE people  (    id BIGINT IDENTITY NOT NULL PRIMARY KEY,    first_name VARCHAR(20),    last_name VARCHAR(20));

3、创建 Person 类

@Datapublic class Person {    private Long id;    private String lastName;    private String firstName;}

4、创建一个中间处理器 PersonItemProcessor

import org.springframework.batch.item.ItemProcessor;@Log4j2public class PersonItemProcessor implements ItemProcessor<Person, Person> {    @Override    public Person process(final Person person) throws Exception {        final String firstName = person.getFirstName().toUpperCase();        final String lastName = person.getLastName().toUpperCase();        final Person transformedPerson = new Person(firstName, lastName);        log.info("Converting (" + person + ") into (" + transformedPerson + ")");        return transformedPerson;    }}

5、创建 PersonMapper，用户数据库映射

public class PersonMapper implements RowMapper {    private static final String ID_COLUMN = "id";    private static final String NICKNAME_COLUMN = "first_name";    private static final String EMAIL_COLUMN = "last_name";    @Override    public Object mapRow(ResultSet resultSet, int i) throws SQLException {        Person user = new Person();        person.setId(resultSet.getLong(ID_COLUMN));        person.setNickname(resultSet.getString(NICKNAME_COLUMN));        person.setEmail(resultSet.getString(EMAIL_COLUMN));        return person;    }}

6、创建任务完成的监听 JobCompletionNotificationListener

@Log4j2@Componentpublic class JobCompletionNotificationListener extends JobExecutionListenerSupport {    @Override    public void afterJob(JobExecution jobExecution) {        if(jobExecution.getStatus() == BatchStatus.COMPLETED) {            log.info("!!! JOB FINISHED! Time to verify the results");        }    }}

7、构建批处理任务 BatchConfiguration

@Configuration@EnableBatchProcessingpublic class BatchConfiguration {    @Autowired    public JobBuilderFactory jobBuilderFactory;    @Autowired    public StepBuilderFactory stepBuilderFactory;    @Autowired    public DataSource dataSource;        @Autowired    public MongoTemplate mongoTemplate;    @Bean    public JdbcCursorItemReader<Person> reader(){        JdbcCursorItemReader<Person> itemReader = new JdbcCursorItemReader<Person>();        itemReader.setDataSource(dataSource);        itemReader.setSql("select id, nickname, email from people");        itemReader.setRowMapper(new PersonMapper());        return itemReader;    }    @Bean    public PersonItemProcessor processor() {        return new PersonItemProcessor();    }        @Bean    MongoItemWriter<Person> writer(){        MongoItemWriter<Person> itemWriter = new MongoItemWriter<Person>();        itemWriter.setTemplate(mongoTemplate);        itemWriter.setCollection("branch");        return itemWriter;    }    @Bean    public Step step() {        return stepBuilderFactory.get("step")                .<Person, Person> chunk(10)                .reader(reader())                .processor(processor())                .writer(writer())                .build();    }    @Bean    public Job importUserJob(JobCompletionNotificationListener listener) {        return jobBuilderFactory.get("importUserJob")                .incrementer(new RunIdIncrementer())                .listener(listener)                .flow(step())                .end()                .build();    }}