原文： https://howtodoinjava.com/spring-batch/csv-to-database-java-config-example/

学习使用 Spring Batch 从 CSV 文件读取记录，并使用JdbcBatchItemWriter插入数据库。 我正在使用嵌入式数据库 H2 演示此示例。

项目概况

在此应用程序中，我们将执行以下任务：

  1. 使用FlatFileItemReader从 CSV 文件读取员工记录
  2. 配置 H2 数据库并在其中创建EMPLOYEE表
  3. 用JdbcBatchItemWriter将员工记录写入EMPLOYEE表
  4. 使用ItemProcessor将日志项插入数据库
  5. 使用 H2 控制台验证插入的记录

项目结构

Spring Batch CSV 到数据库 – Java 注解配置示例 - 图1

包结构

Maven 依赖

快速浏览构建此示例所需的 maven 依赖关系。 需要spring-boot-starter-web才能从浏览器窗口验证 H2 控制台中的数据。

pom.xml

  1. <project xmlns="http://maven.apache.org/POM/4.0.0"
  2.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.     xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd;
  4.     <modelVersion>4.0.0</modelVersion>
  6.     <groupId>com.howtodoinjava</groupId>
  7.     <artifactId>App</artifactId>
  8.     <version>0.0.1-SNAPSHOT</version>
  9.     <packaging>jar</packaging>
  11.     <name>App</name>
  12.     <url>http://maven.apache.org</url>
  14.     <parent>
  15.         <groupId>org.springframework.boot</groupId>
  16.         <artifactId>spring-boot-starter-parent</artifactId>
  17.         <version>2.0.3.RELEASE</version>
  18.     </parent>
  20.     <properties>
  21.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  22.     </properties>
  24.     <dependencies>
  25.         <dependency>
  26.             <groupId>org.springframework.boot</groupId>
  27.             <artifactId>spring-boot-starter-batch</artifactId>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>org.springframework.boot</groupId>
  31.             <artifactId>spring-boot-starter-web</artifactId>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>com.h2database</groupId>
  35.             <artifactId>h2</artifactId>
  36.             <scope>runtime</scope>
  37.         </dependency>
  38.     </dependencies>
  40.     <build>
  41.         <plugins>
  42.             <plugin>
  43.                 <groupId>org.springframework.boot</groupId>
  44.                 <artifactId>spring-boot-maven-plugin</artifactId>
  45.             </plugin>
  46.         </plugins>
  47.     </build>
  49.     <repositories>
  50.         <repository>
  51.             <id>repository.spring.release</id>
  52.             <name>Spring GA Repository</name>
  53.             <url>http://repo.spring.io/release</url>
  54.         </repository>
  55.     </repositories>
  56. </project>

CSV 读取器和数据库写入器配置

  1. 我们将使用FlatFileItemReader读取 CSV 文件。 我们将使用涉及DefaultLineMapper，DelimitedLineTokenizer和BeanWrapperFieldSetMapper类的标准配置。
  2. 为了将记录写入数据库，我们将使用JdbcBatchItemWriter这是标准编写器，用于在数据库中为 Spring Batch 作业执行批处理查询。

BatchConfig.java

  1. package com.howtodoinjava.demo.config;
  3. import javax.sql.DataSource;
  4. import org.springframework.batch.core.Job;
  5. import org.springframework.batch.core.Step;
  6. import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
  7. import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
  8. import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
  9. import org.springframework.batch.core.launch.support.RunIdIncrementer;
  10. import org.springframework.batch.item.ItemProcessor;
  11. import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
  12. import org.springframework.batch.item.database.JdbcBatchItemWriter;
  13. import org.springframework.batch.item.file.FlatFileItemReader;
  14. import org.springframework.batch.item.file.LineMapper;
  15. import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
  16. import org.springframework.batch.item.file.mapping.DefaultLineMapper;
  17. import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
  18. import org.springframework.beans.factory.annotation.Autowired;
  19. import org.springframework.beans.factory.annotation.Value;
  20. import org.springframework.context.annotation.Bean;
  21. import org.springframework.context.annotation.Configuration;
  22. import org.springframework.core.io.Resource;
  23. import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
  24. import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
  25. import com.howtodoinjava.demo.model.Employee;
  27. @Configuration
  28. @EnableBatchProcessing
  29. public class BatchConfig {
  31.     @Autowired
  32.     private JobBuilderFactory jobBuilderFactory;
  34.     @Autowired
  35.     private StepBuilderFactory stepBuilderFactory;
  37.     @Value("classPath:/input/inputData.csv")
  38.     private Resource inputResource;
  40.     @Bean
  41.     public Job readCSVFileJob() {
  42.         return jobBuilderFactory
  43.                 .get("readCSVFileJob")
  44.                 .incrementer(new RunIdIncrementer())
  45.                 .start(step())
  46.                 .build();
  47.     }
  49.     @Bean
  50.     public Step step() {
  51.         return stepBuilderFactory
  52.                 .get("step")
  53.                 .<Employee, Employee>chunk(5)
  54.                 .reader(reader())
  55.                 .processor(processor())
  56.                 .writer(writer())
  57.                 .build();
  58.     }
  60.     @Bean
  61.     public ItemProcessor<Employee, Employee> processor() {
  62.         return new DBLogProcessor();
  63.     }
  65.     @Bean
  66.     public FlatFileItemReader<Employee> reader() {
  67.         FlatFileItemReader<Employee> itemReader = new FlatFileItemReader<Employee>();
  68.         itemReader.setLineMapper(lineMapper());
  69.         itemReader.setLinesToSkip(1);
  70.         itemReader.setResource(inputResource);
  71.         return itemReader;
  72.     }
  74.     @Bean
  75.     public LineMapper<Employee> lineMapper() {
  76.         DefaultLineMapper<Employee> lineMapper = new DefaultLineMapper<Employee>();
  77.         DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
  78.         lineTokenizer.setNames(new String[] { "id", "firstName", "lastName" });
  79.         lineTokenizer.setIncludedFields(new int[] { 0, 1, 2 });
  80.         BeanWrapperFieldSetMapper<Employee> fieldSetMapper = new BeanWrapperFieldSetMapper<Employee>();
  81.         fieldSetMapper.setTargetType(Employee.class);
  82.         lineMapper.setLineTokenizer(lineTokenizer);
  83.         lineMapper.setFieldSetMapper(fieldSetMapper);
  84.         return lineMapper;
  85.     }
  87.     @Bean
  88.     public JdbcBatchItemWriter<Employee> writer() {
  89.         JdbcBatchItemWriter<Employee> itemWriter = new JdbcBatchItemWriter<Employee>();
  90.         itemWriter.setDataSource(dataSource());
  91.         itemWriter.setSql("INSERT INTO EMPLOYEE (ID, FIRSTNAME, LASTNAME) VALUES (:id, :firstName, :lastName)");
  92.         itemWriter.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Employee>());
  93.         return itemWriter;
  94.     }
  96.     @Bean
  97.     public DataSource dataSource(){
  98.         EmbeddedDatabaseBuilder embeddedDatabaseBuilder = new EmbeddedDatabaseBuilder();
  99.         return embeddedDatabaseBuilder.addScript("classpath:org/springframework/batch/core/schema-drop-h2.sql")
  100.                 .addScript("classpath:org/springframework/batch/core/schema-h2.sql")
  101.                 .addScript("classpath:employee.sql")
  102.                 .setType(EmbeddedDatabaseType.H2)
  103.                 .build();
  104.     }
  105. }

还创建DBLogProcessor，它将在写入数据库之前记录员工记录。 它是可选的。

DBLogProcessor.java

  1. package com.howtodoinjava.demo.config;
  3. import org.springframework.batch.item.ItemProcessor;
  4. import com.howtodoinjava.demo.model.Employee;
  6. public class DBLogProcessor implements ItemProcessor<Employee, Employee>
  7. {
  8.     public Employee process(Employee employee) throws Exception
  9.     {
  10.         System.out.println("Inserting employee : " + employee);
  11.         return employee;
  12.     }
  13. }

模型类

Employee.java

  1. package com.howtodoinjava.demo.config;
  2. package com.howtodoinjava.demo.model;
  4. public class Employee {
  6.     String id;
  7.     String firstName;
  8.     String lastName;
  10.     //Setter and getter methods
  11. }

应用属性

application.properties

  1. #Disable batch job's auto start 
  2. spring.batch.job.enabled=false
  3. spring.main.banner-mode=off
  5. #batch input files location
  6. input.dir=c:/temp/input

日志配置

logback.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration scan="true">
  4.     <appender name="consoleAppender" class="ch.qos.logback.core.ConsoleAppender">
  5.         <encoder>
  6.             <charset>UTF-8</charset>
  7.             <Pattern>%d{yyyy-MM-dd HH:mm:ss} %p %X{TXNID} - %m%n</Pattern>
  8.         </encoder>
  9.     </appender>
  11.     <root level="INFO">
  12.         <appender-ref ref="consoleAppender" />
  13.     </root>
  14. </configuration>

配置 H2 数据库

我们已经在BatchConfig.java中配置了数据源。

BatchConfig.java

  1. @Bean
  2. public DataSource dataSource(){
  3.     EmbeddedDatabaseBuilder embeddedDatabaseBuilder = new EmbeddedDatabaseBuilder();
  4.     return embeddedDatabaseBuilder.addScript("classpath:org/springframework/batch/core/schema-drop-h2.sql")
  5.             .addScript("classpath:org/springframework/batch/core/schema-h2.sql")
  6.             .addScript("classpath:employee.sql")
  7.             .setType(EmbeddedDatabaseType.H2)
  8.             .build();
  9. }

创建EMPLOYEE表

以上配置将自动生成默认表。 要生成EMPLOYEE表，请创建模式文件employee.sql并将其放置在resources文件夹中。

employee.sql

  1. DROP TABLE EMPLOYEE IF EXISTS;
  3. CREATE TABLE EMPLOYEE  (
  4.     ID VARCHAR(10),  
  5.     FIRSTNAME VARCHAR(100),  
  6.     LASTNAME VARCHAR(100) 
  7. ) ;

启用 H2 控制台

要启用 H2 控制台，请向 Spring Web 注册org.h2.server.web.WebServlet。

WebConfig.java

  1. package com.howtodoinjava.demo.config;
  3. import org.h2.server.web.WebServlet;
  4. import org.springframework.boot.web.servlet.ServletRegistrationBean;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. @SuppressWarnings({"rawtypes","unchecked"})
  9. @Configuration
  10. public class WebConfig {
  12.     @Bean
  13.     ServletRegistrationBean h2servletRegistration(){
  14.         ServletRegistrationBean registrationBean = new ServletRegistrationBean( new WebServlet());
  15.         registrationBean.addUrlMappings("/console/*");
  16.         return registrationBean;
  17.     }
  18. }

示例

我们的应用程序配置完成，作业准备执行。 让我们创建输入 CSV 文件。

inputData.csv

  1. id,firstName,lastName
  2. 1,Lokesh,Gupta
  3. 2,Amit,Mishra
  4. 3,Pankaj,Kumar
  5. 4,David,Miller
  6. 5,David,Walsh

运行演示

要运行演示和批处理作业，请创建 Spring 运行应用程序类并启动应用程序。

App.java

  1. package com.howtodoinjava.demo;
  3. import org.springframework.batch.core.Job;
  4. import org.springframework.batch.core.JobParameters;
  5. import org.springframework.batch.core.JobParametersBuilder;
  6. import org.springframework.batch.core.launch.JobLauncher;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.SpringApplication;
  9. import org.springframework.boot.autoconfigure.SpringBootApplication;
  10. import org.springframework.scheduling.annotation.EnableScheduling;
  11. import org.springframework.scheduling.annotation.Scheduled;
  13. @SpringBootApplication
  14. @EnableScheduling
  15. public class App
  16. {
  17.     @Autowired
  18.     JobLauncher jobLauncher;
  20.     @Autowired
  21.     Job job;
  23.     public static void main(String[] args)
  24.     {
  25.         SpringApplication.run(App.class, args);
  26.     }
  28.     @Scheduled(cron = "0 */1 * * * ?")
  29.     public void perform() throws Exception
  30.     {
  31.         JobParameters params = new JobParametersBuilder()
  32.                 .addString("JobID", String.valueOf(System.currentTimeMillis()))
  33.                 .toJobParameters();
  34.         jobLauncher.run(job, params);
  35.     }
  36. }

验证批处理作业结果

要验证批处理作业是否成功执行，请检查日志和 H2 控制台。

Console

  1. 2018-07-11 19:11:00 INFO  - Job: [SimpleJob: [name=readCSVFileJob]] launched with the following parameters: [{JobID=1531316460004}]
  3. 2018-07-11 19:11:00 INFO  - Executing step: [step]
  5. Inserting employee : Employee [id=1, firstName=Lokesh, lastName=Gupta]
  6. Inserting employee : Employee [id=2, firstName=Amit, lastName=Mishra]
  7. Inserting employee : Employee [id=3, firstName=Pankaj, lastName=Kumar]
  8. Inserting employee : Employee [id=4, firstName=David, lastName=Miller]
  9. Inserting employee : Employee [id=5, firstName=David, lastName=Walsh]
  11. 2018-07-11 19:11:00 INFO  - Job: [SimpleJob: [name=readCSVFileJob]] completed with the following parameters: [{JobID=1531316460004}] and the following status: [COMPLETED]

H2 控制台

Spring Batch CSV 到数据库 – Java 注解配置示例 - 图2

H2 控制台登录

Spring Batch CSV 到数据库 – Java 注解配置示例 - 图3

H2 控制台中的数据

将我的问题放在评论部分。

学习愉快！

下载源码