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

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

2019 年 10 月 30 日
筆記

从 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();      }    }