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

    <!-- 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>

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

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

import org.springframework.batch.item.ItemProcessor;@Log4j2public 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;  }}

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;  }}

@Log4j2@Componentpublic class JobCompletionNotificationListener extends JobExecutionListenerSupport {  @Override  public void afterJob(JobExecution jobExecution) {    if(jobExecution.getStatus() == BatchStatus.COMPLETED) {      log.info("!!! JOB FINISHED! Time to verify the results");    }  }}

@Configuration@EnableBatchProcessingpublic 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();  }}