Spring Boot 集成批处理框架Spring batch

╰+哭是因爲堅強的太久メ 2022-08-19 15:17 41阅读 0赞

**Spring Batch是一个轻量级的框架,完全面向Spring的批处理框架,用于企业级大量的数据读写处理系统。以POJO和Spring 框架为基础，包括日志记录/跟踪，事务管理、 作业处理统计工作重新启动、跳过、资源管理等功能。**

**       业务方案：**

**1、批处理定期提交。**

**2、并行批处理：并行处理工作。**

**3、企业消息驱动处理**

**4、大规模的并行处理**

**5、手动或是有计划的重启**

**6、局部处理：跳过记录（如：回滚）**

**      技术目标：**

**1、利用Spring编程模型：使程序员专注于业务处理，让Spring框架管理流程。**

**2、明确分离批处理的执行环境和应用。**

**3、提供核心的，共通的接口。**

**4、提供开箱即用（out of the box）的简单的默认的核心执行接口。**

**5、提供Spring框架中配置、自定义、和扩展服务。**

**6、所有存在的核心服务可以很容的被替换和扩展，不影响基础层。**

**7、提供一个简单的部署模式，利用Maven构建独立的Jar文件。**

**实现步骤:**

**1.定义处理对象**

**2.创建中间转换器 \*\*\*ItemProcessor 实现 ItemProcessor<I,O>接口**

**3.创建工作Job BatchConfiguration 主要处理读数据、处理数据、写数据等操作**

**4.创建listener job执行监听器**

** **

**一般的批处理系统需要处理大量的数据, 内部消化单条记录失败的情况, 还要管理中断,在重启后也不去重复执行已经处理过的部分**

** **

** Spring Batch单/多处理单元(processors), 以及多个微线程(tasklets)**

具体实现过程:

1.引入jar:

<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>1.3.1.RELEASE</version>
    	</parent>
    	<properties>
    		<java.version>1.8</java.version>
    	</properties>
    
        <!--Spring batch核心包->
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-batch</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>mysql</groupId>
    			<artifactId>mysql-connector-java</artifactId>
    		</dependency>
      
    </dependencies>

2.配置application.properties ：配置数据源

######mysql\u6570\u636E\u6E90#########
    spring.datasource.driver-class-name=com.mysql.jdbc.Driver
    spring.datasource.password= wbw123456
    spring.datasource.url= jdbc:mysql://localhost/mydatabase
    spring.datasource.username=root
    <a target=_blank href="http://projects.spring.io/spring-batch/">点击打开链接</a>

3.新建实体类

package com.my.gs.batch.processing.domain;
    
    public class Person {
    	//ID
        private Integer personId;
        //姓名
        private String personName;
        //年龄
        private String personAge;
        //性别
        private String personSex;
        
        public Person(){};
    	public Person( String personName, String personAge,
    			String personSex) {
    		this.personName = personName;
    		this.personAge = personAge;
    		this.personSex = personSex;
    	}
    	public Integer getPersonId() {
    		return personId;
    	}
    	public void setPersonId(Integer personId) {
    		this.personId = personId;
    	}
    	public String getPersonName() {
    		return personName;
    	}
    	public void setPersonName(String personName) {
    		this.personName = personName;
    	}
    	public String getPersonAge() {
    		return personAge;
    	}
    	public void setPersonAge(String personAge) {
    		this.personAge = personAge;
    	}
    	public String getPersonSex() {
    		return personSex;
    	}
    	public void setPersonSex(String personSex) {
    		this.personSex = personSex;
    	}
    
        
        
    }

4.中间转换器：

package com.my.gs.batch.processing.itemprocessor;
    
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.util.List;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.batch.item.ItemProcessor;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.jdbc.core.RowMapper;
    
    import com.my.gs.batch.processing.domain.Person;
    /**
     * 中间转换器
     * @author wbw
     *
     */
    public class PersonItemProcessor implements ItemProcessor<Person, Person> {
    	//查询
    	private static final String GET_PRODUCT = "select * from Person where personName = ?";
    	
        private static final Logger log = LoggerFactory.getLogger(PersonItemProcessor.class);
        @Autowired
        private JdbcTemplate jdbcTemplate;
        @Override
        public Person process(final Person person) throws Exception {
            List<Person> personList = jdbcTemplate.query(GET_PRODUCT, new Object[] {person.getPersonName()}, new RowMapper<Person>() {
                @Override
                public Person mapRow( ResultSet resultSet, int rowNum ) throws SQLException {
                	Person p = new Person();
                	p.setPersonName(resultSet.getString(1));
                	p.setPersonAge(resultSet.getString(2));
                	p.setPersonSex(resultSet.getString(3));
                    return p;
                }
            });
            if(personList.size() >0){
            	log.info("该数据已录入!!!");
            }
        	String sex = null;
            if(person.getPersonSex().equals("0")){
            	sex ="男";
            }else{
            	sex ="女";
            }
            log.info("转换 (性别："+person.getPersonSex()+") 为 (" + sex + ")");
            final Person transformedPerson = new Person(person.getPersonName(), person.getPersonAge(),sex);
            log.info("转换 (" + person + ") 为 (" + transformedPerson + ")");
    
            return transformedPerson;
        }
    
    }

5.处理具体工作业务  主要包含三个部分:读数据、处理数据、写数据

package com.my.gs.batch.processing.configuration;
    
    import javax.sql.DataSource;
    
    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.ItemProcessor;
    import org.springframework.batch.item.ItemReader;
    import org.springframework.batch.item.ItemWriter;
    import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
    import org.springframework.batch.item.database.JdbcBatchItemWriter;
    import org.springframework.batch.item.file.FlatFileItemReader;
    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.Qualifier;
    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 com.my.gs.batch.processing.domain.Person;
    import com.my.gs.batch.processing.itemprocessor.PersonItemProcessor;
    /**
     * 处理具体工作业务  主要包含三个部分:读数据、处理数据、写数据
     * @author wbw
     *
     */
    @Configuration
    @EnableBatchProcessing
    public class PersonBatchConfiguration {
    	//插入语句
       private static final String PERSON_INSERT = "INSERT INTO Person (personName, personAge,personSex) VALUES (:personName, :personAge,:personSex)";
       public static final String Person_INSERT = "INSERT INTO Person (id, name,description,quantity) VALUES (:id, :name,:description,:quantity)";
       // tag::readerwriterprocessor[] 1.读数据
        @Bean
        public ItemReader<Person> reader() {
            FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>();
            //加载外部文件数据 文件类型:CSV
            reader.setResource(new ClassPathResource("sample-data.csv"));
            reader.setLineMapper(new DefaultLineMapper<Person>() {
       {
                setLineTokenizer(new DelimitedLineTokenizer() {
       {
                    setNames(new String[] { "personName","personAge","personSex" });
                }});
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
       {
                    setTargetType(Person.class);
                }});
            }});
            return reader;
        }
        //2.处理数据
        @Bean
        public PersonItemProcessor processor() {
            return new PersonItemProcessor();
        }
        //3.写数据
        @Bean
        public ItemWriter<Person> writer(DataSource dataSource) {
            JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<Person>();
            writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
            writer.setSql(PERSON_INSERT);
            writer.setDataSource(dataSource);
            return writer;
        }
        // end::readerwriterprocessor[]
    
        
        // tag::jobstep[]
        @Bean
        public Job importUserJob(JobBuilderFactory jobs, @Qualifier("step1")Step s1, JobExecutionListener listener) {
            return jobs.get("importUserJob")
                    .incrementer(new RunIdIncrementer())
                    .listener(listener)
                    .flow(s1)
                    .end()
                    .build();
        }
    
        @Bean
        public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader,
                ItemWriter<Person> writer, ItemProcessor<Person, Person> processor) {
            return stepBuilderFactory.get("step1")
                    .<Person, Person> chunk(10)
                    .reader(reader)
                    .processor(processor)
                    .writer(writer)
                    .build();
        }
        // end::jobstep[]
    
        @Bean
        public JdbcTemplate jdbcTemplate(DataSource dataSource) {
            return new JdbcTemplate(dataSource);
        }
    
    }

6.监听器:用于处理任务执行之后和之前

<pre name="code" class="java">package com.my.gs.batch.processing.listener;
    
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.util.List;
    
    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.jdbc.core.RowMapper;
    import org.springframework.stereotype.Component;
    
    import com.my.gs.batch.processing.domain.Person;
    /**
     * Job执行监听器
     * @author wbw
     *
     */
    @Component
    public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
       
       private static final String PERSON_SQL = "SELECT personName, personAge,personSex FROM Person";
    	
       private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);
        @Autowired
    	private  JdbcTemplate jdbcTemplate;
    
    	@Autowired
    	public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {
    		if(this.jdbcTemplate==null){
    		this.jdbcTemplate = jdbcTemplate;
    		}
    	}
    
    	@Override
    	public void afterJob(JobExecution jobExecution) {
    		if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
    			log.info("!!! JOB 执行完成!");
    			List<Person> results = jdbcTemplate.query(PERSON_SQL, new RowMapper<Person>() {
    				@Override
    				public Person mapRow(ResultSet rs, int row) throws SQLException {
    					return new Person(rs.getString(1), rs.getString(2),rs.getString(3));
    				}
    			});
                log.info("入库条数---------"+results.size());
    			for (Person person : results) {
    				log.info("新增 <" + person.getPersonName() + "> 成功!!!!!");
    			}
    
    		}
    	}
    
    	/* (non-Javadoc)
    	 * @see org.springframework.batch.core.listener.JobExecutionListenerSupport#beforeJob(org.springframework.batch.core.JobExecution)
    	 */
    	@Override
    	public void beforeJob(JobExecution jobExecution) {
    		// TODO Auto-generated method stub
    		super.beforeJob(jobExecution);
    	}
    	
    }

7.新建csv文件

</pre><pre code_snippet_id="1563739" snippet_file_name="blog_20160124_10_8470765" name="code" class="java">8.启动执行

<pre name="code" class="java">package com.my.batch;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class Application {
    
        public static void main(String[] args) throws Exception {
            SpringApplication.run(Application.class, args);
        }
    }

更多相关资源:http://projects.spring.io/spring-batch/