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

从 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 类

@Data

public class Person {

    private Long id;

    private String lastName;

    private String firstName;

}

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

import org.springframework.batch.item.ItemProcessor;

@Log4j2

public 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

@Component

public 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

@EnableBatchProcessing

public 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();

    }

}