Spring Boot 之 Spring Batch 批处理实践

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

具体实现

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(); } } 任务处理结果

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