實踐內(nèi)容

從 MariaDB 一張表內(nèi)讀 10 萬條記錄，經(jīng)處理后寫到 MongoDB 囊颅。

Batch 任務(wù)模型

具體實現(xiàn)

1、新建 Spring Boot 應(yīng)用傅瞻，依賴如下：

        <!-- Web 應(yīng)用 -->
        <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 驅(qū)動 -->
        <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踢代、創(chuàng)建一張表，并生成 10 萬條數(shù)據(jù)

DROP TABLE people IF EXISTS;

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

3嗅骄、創(chuàng)建 Person 類

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

4胳挎、創(chuàng)建一個中間處理器 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、創(chuàng)建 PersonMapper溺森，用戶數(shù)據(jù)庫映射

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慕爬、創(chuàng)建任務(wù)完成的監(jiān)聽 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、構(gòu)建批處理任務(wù) 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();
    }

}

任務(wù)處理結(jié)果

0出錯儿惫，耗時 2 分鐘左右澡罚，測試機 Mac