详解Spring Batch 轻量级批处理框架实践

实践内容

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

具体实现

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

    
    
      org.springframework.boot
      spring-boot-starter-web
      
 
   org.springframework.boot
   spring-boot-starter-logging
 
 
   org.springframework.boot
   spring-boot-starter-tomcat
 
      
    

    
    
      org.springframework.boot
      spring-boot-starter-undertow
    

    
    
      org.springframework.boot
      spring-boot-starter-log4j2
    

    
    
      org.springframework.boot
      spring-boot-starter-data-mongodb
    

    
    
      org.springframework.boot
      spring-boot-starter-batch
    

    
    
      org.mariadb.jdbc
      mariadb-java-client
      2.0.2
    

    
    
      org.projectlombok
      lombok
      1.16.14
    

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 {

  @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 reader(){
    JdbcCursorItemReader itemReader = new JdbcCursorItemReader();
    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 writer(){
    MongoItemWriter itemWriter = new MongoItemWriter();
    itemWriter.setTemplate(mongoTemplate);
    itemWriter.setCollection("branch");
    return itemWriter;
  }

  @Bean
  public Step step() {
    return stepBuilderFactory.get("step")
 . 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();
  }

}

任务处理结果

0出错，耗时 2 分钟左右，测试机 Mac

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持考高分网。