Spring实现批处理任务

心已赠人 2024-04-18 08:45 59阅读 0赞

### Spring实现批处理任务 ###

*  前言
 *  创建业务数据
 *  编写SQL脚本
 *  创建实体类
 *  创建一个中间处理器

# 前言 #

本文主要讲述基本的批处理任务的解决方案。

# 创建业务数据 #

手动创建一个文件来模拟业务数据  
src/main/resources/sample-data.properties

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

# 编写SQL脚本 #

接下来写一个sql脚本创建一个表来存储数据  
src/main/resources/schema-all.sql

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

> Spring Boot运行模式-@@platform@@ @。sql在启动时自动执行。-all是所有平台的默认值。

# 创建实体类 #

接下来写一个类来表示每行的数据

package com.yjn.bean;
    
    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.yjn.batch;
    
    import com.yjn.bean.Person;
    import org.springframework.batch.core.Job;
    import org.springframework.batch.core.JobExecutionListener;
    import org.springframework.batch.core.Step;
    import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
    import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
    import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
    import org.springframework.batch.core.launch.support.RunIdIncrementer;
    import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
    import org.springframework.batch.item.database.JdbcBatchItemWriter;
    import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
    import org.springframework.batch.item.file.FlatFileItemReader;
    import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
    import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
    import org.springframework.batch.item.file.mapping.DefaultLineMapper;
    import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.core.io.ClassPathResource;
    import org.springframework.jdbc.core.JdbcTemplate;
    
    import javax.sql.DataSource;
    
    @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();
        }
    }

对于初学者来说，@EnableBatchProcessing注释添加了很多任务支持的关键bean，并节省了大量的辅助工作。这个例子使用一个基于内存的数据库(由@EnableBatchProcessing提供)，这意味着当它完成时，数据就消失了。  
第一个代码块定义输入、处理器和输出。  
  reader()创建一个ItemReader。它查找一个名为sample-data.csv的文件，并解析每个行项目，其中包含足够的信息将其转换为Person。  
processor()创建前面定义的PersonItemProcessor的一个实例，用于大写数据。  
write(DataSource)创建一个ItemWriter。这一个针对JDBC目标，并自动获取由@EnableBatchProcessing创建的数据源的副本。它包括插入由Java bean属性驱动的单个人所需的SQL语句。

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

第一个方法定义作业，第二个方法定义单个步骤。作业是由步骤构建的，其中每个步骤都包含一个读取器、一个处理器和一个写入器。  
在这个作业定义中，你需要一个递增器，因为作业使用数据库来维护执行状态。然后列出每个步骤，其中这个作业只有一个步骤。作业结束，Java API生成一个配置完美的作业。  
在步骤定义中，定义一次要写多少数据。在本例中，它一次写入多达10条记录。接下来，使用前面注入的位配置读取器、处理器和写入器。

package com.yjn.batch;
    
    import com.yjn.bean.Person;
    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."));
            }
        }
    }