创建批处理服务

Jill,Doe
Joe,Doe
Justin,Doe
Jane,Doe
John,Doe

DROP TABLE people IF EXISTS;

CREATE TABLE people  (
    person_id BIGINT IDENTITY NOT NULL PRIMARY KEY,
    first_name VARCHAR(20),
    last_name VARCHAR(20)
);

<？xml版本=“ 1.0”编码=“ UTF-8”？>
<project xmlns =“ http://maven.apache.org/POM/4.0.0” xmlns：xsi =“ http://www.w3.org/2001/XMLSchema-instance”
	xsi：schemaLocation =“ http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd”>
	<modelVersion> 4.0.0 </ modelVersion>
	<父母>
		<groupId> org.springframework.boot </ groupId>
		<artifactId> spring-boot-starter-parent </ artifactId>
		<version> 2.4.3 </ version>
		<relativePath /> <！-从存储库中查找父级->
	</ parent>
	<groupId> com.example </ groupId>
	<artifactId>批处理</ artifactId>
	<version> 0.0.1-SNAPSHOT </ version>
	<名称>批处理</名称>
	<description> Spring Boot的演示项目</ description>
	<属性>
		<java.version> 1.8 </java.version>
	</ properties>
	<依赖项>
		<依赖性>
			<groupId> org.springframework.boot </ groupId>
			<artifactId> spring-boot-starter-batch </ artifactId>
		</ dependency>

		<依赖性>
			<groupId> org.hsqldb </ groupId>
			<artifactId> hsqldb </ artifactId>
			<scope>运行时</ scope>
		</ dependency>
		<依赖性>
			<groupId> org.springframework.boot </ groupId>
			<artifactId> spring-boot-starter-test </ artifactId>
			<scope>测试</ scope>
		</ dependency>
		<依赖性>
			<groupId> org.springframework.batch </ groupId>
			<artifactId> spring-batch-test </ artifactId>
			<scope>测试</ scope>
		</ dependency>
	</ dependencies>

	<内部版本>
		<插件>
			<插件>
				<groupId> org.springframework.boot </ groupId>
				<artifactId> spring-boot-maven-plugin </ artifactId>
			</ plugin>
		</ plugins>
	</ build>

</ project>

插件{
	id'org.springframework.boot'版本'2.4.3'
	id'io.spring.dependency-management'版本'1.0.11.RELEASE'
	id'java'
}

组='com.example'
版本='0.0.1-SNAPSHOT'
sourceCompatibility ='1.8'

储存库{
	mavenCentral（）
}

依赖项{
	实现'org.springframework.boot：spring-boot-starter-batch'
	runtime仅'org.hsqldb：hsqldb'
	testImplementation'org.springframework.boot：spring-boot-starter-test'
	testImplementation'org.springframework.batch：spring-batch-test'
}

测试 {
	useJUnitPlatform（）
}

package com.example.batchprocessing;

public class Person {

  private String lastName;
  private String firstName;

  public Person() {
  }

  public Person(String firstName, String lastName) {
    this.firstName = firstName;
    this.lastName = lastName;
  }

  public void setFirstName(String firstName) {
    this.firstName = firstName;
  }

  public String getFirstName() {
    return firstName;
  }

  public String getLastName() {
    return lastName;
  }

  public void setLastName(String lastName) {
    this.lastName = lastName;
  }

  @Override
  public String toString() {
    return "firstName: " + firstName + ", lastName: " + lastName;
  }

}

package com.example.batchprocessing;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.batch.item.ItemProcessor;

public class PersonItemProcessor implements ItemProcessor<Person, Person> {

  private static final Logger log = LoggerFactory.getLogger(PersonItemProcessor.class);

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

}

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

  @Autowired
  public JobBuilderFactory jobBuilderFactory;

  @Autowired
  public StepBuilderFactory stepBuilderFactory;

    ...

}

@Bean
public FlatFileItemReader<Person> reader() {
  return new FlatFileItemReaderBuilder<Person>()
    .name("personItemReader")
    .resource(new ClassPathResource("sample-data.csv"))
    .delimited()
    .names(new String[]{"firstName", "lastName"})
    .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
      setTargetType(Person.class);
    }})
    .build();
}

@Bean
public PersonItemProcessor processor() {
  return new PersonItemProcessor();
}

@Bean
public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
  return new JdbcBatchItemWriterBuilder<Person>()
    .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
    .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
    .dataSource(dataSource)
    .build();
}

@Bean
public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
  return jobBuilderFactory.get("importUserJob")
    .incrementer(new RunIdIncrementer())
    .listener(listener)
    .flow(step1)
    .end()
    .build();
}

@Bean
public Step step1(JdbcBatchItemWriter<Person> writer) {
  return stepBuilderFactory.get("step1")
    .<Person, Person> chunk(10)
    .reader(reader())
    .processor(processor())
    .writer(writer)
    .build();
}

package com.example.batchprocessing;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

@Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {

  private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);

  private final JdbcTemplate jdbcTemplate;

  @Autowired
  public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {
    this.jdbcTemplate = jdbcTemplate;
  }

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

      jdbcTemplate.query("SELECT first_name, last_name FROM people",
        (rs, row) -> new Person(
          rs.getString(1),
          rs.getString(2))
      ).forEach(person -> log.info("Found <" + person + "> in the database."));
    }
  }
}

package com.example.batchprocessing;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class BatchProcessingApplication {

  public static void main(String[] args) throws Exception {
    System.exit(SpringApplication.exit(SpringApplication.run(BatchProcessingApplication.class, args)));
  }
}