快捷導(dǎo)航

詳解Spring Batch 輕量級批處理框架實踐

更新時間：2019年06月27日 14:53:43 作者：Anoy

這篇文章主要介紹了詳解Spring Batch 輕量級批處理框架實踐，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

實踐內(nèi)容

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

具體實現(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

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: