Practice content

Read 100,000 records from a MariaDB table and write them to MongoDB after processing.

The specific implementation

1. Create a Spring Boot application with the following dependencies:

<! -- Web application --> <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> <! <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-undertow</artifactId> </dependency> <! <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 driver --> <dependency> <groupId>org.mariadb. JDBC </groupId> <artifactId> Mariadb -java-client</artifactId> The < version > 2.0.2 < / version > < / dependency > <! --> <dependency> <groupId>org.projectlombok</groupId> <artifactId> Lombok </artifactId> The < version > 1.16.14 < / version > < / dependency >Copy the code

Create a table and generate 100,000 pieces of data

DROP TABLE people IF EXISTS;

CREATE TABLE people  (
    id BIGINT IDENTITY NOT NULL PRIMARY KEY,
    first_name VARCHAR(20),
    last_name VARCHAR(20)
);Copy the code

Create the Person class

@Data
public class Person {
    private Long id;
    private String lastName;
    private String firstName;
}Copy the code

Create an intermediate processor 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; }}Copy the code

Create PersonMapper, user database mapping

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; }}Copy the code

The monitoring JobCompletionNotificationListener 6, creating a task to complete

@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"); }}}Copy the code

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(); }}Copy the code

Task Processing result

0 error, take about 2 minutes, test machine Mac