Spring Boot整合Spring Batch

時間 2019-12-09

標籤 spring boot 整合 batch 欄目 Spring 简体版

原文原文鏈接

引言

　　Spring Batch是處理大量數據操做的一個框架，主要用來讀取大量數據，而後進行必定的處理後輸出指定的形式。好比咱們能夠將csv文件中的數據（數據量幾百萬甚至幾千萬都是沒問題的）批處理插入保存到數據庫中，就可使用該框架，可是不論是數據資料仍是網上資料，我看到不多有這樣的詳細講解。因此本片博文的主要目的邊講解的同時邊實戰（其中的代碼都是通過實踐的）。一樣地先從Spring Boot對Batch框架的支持提及，最後一步一步進行代碼實踐！css

1、Spring Boot對Batch框架的支持

一、Spring Batch框架的組成部分

　　1）JobRepository：用來註冊Job容器，設置數據庫相關屬性。java

　　2）JobLauncher：用來啓動Job的接口mysql

　　3）Job：咱們要實際執行的任務，包含一個或多個spring

　　4）Step：即步驟，包括：ItemReader->ItemProcessor->ItemWritersql

　　5）ItemReader：用來讀取數據，作實體類與數據字段之間的映射。好比讀取csv文件中的人員數據，以後對應實體person的字段作mapper數據庫

　　6）ItemProcessor：用來處理數據的接口，同時能夠作數據校驗（設置校驗器，使用JSR-303(hibernate-validator)註解），好比將中文性別男/女，轉爲M/F。同時校驗年齡字段是否符合要求等springboot

　　7）ItemWriter：用來輸出數據的接口，設置數據庫源。編寫預處理SQL插入語句app

以上七個組成部分，只須要在配置類中逐一註冊便可，同時配置類須要開啓@EnableBatchProcessing註解框架

@Configuration
@EnableBatchProcessing // 開啓批處理的支持
@Import(DruidDBConfig.class) // 注入datasource
public class CsvBatchConfig {
    
}

二、批處理流程圖

以下流程圖便可以解釋在配置類中爲何須要這麼定義，具體請看實戰部分的代碼。ide

2、實戰

一、添加依賴

1）spring batch依賴

<!--  spring batch -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

2）校驗器依賴

<!-- hibernate validator -->
<dependency>
    <groupId>org.hibernate</groupId>
    <artifactId>hibernate-validator</artifactId>
    <version>6.0.7.Final</version>
</dependency>

3）mysql+druid依賴

<!-- mysql connector-->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.35</version>
</dependency>
<!-- alibaba dataSource -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid</artifactId>
    <version>1.1.12</version>
</dependency>

4）test測試依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
</dependency>

二、application.yml配置

當job發佈開始執行任務時，spring batch會自動生成相關的batch開頭的表。這些表一開始是不存在的！須要在application配置文件中作相關的設置。

# batch
  batch:
    job:
      # 默認自動執行定義的Job(true)，改成false，須要jobLaucher.run執行
      enabled: false
    # spring batch在數據庫裏面建立默認的數據表，若是不是always則會提示相關表不存在
    initialize-schema: always
    # 設置batch表的前綴
#    table-prefix: csv-batch

三、數據源配置

  datasource:
    username: root
    password: 1234
    url: jdbc:mysql://127.0.0.1:3306/db_base?useSSL=false&serverTimezone=UTC&characterEncoding=utf8
    driver-class-name: com.mysql.jdbc.Driver

註冊DBConfig配置類：以後經過import導入batch配置類中

/**
 * @author jian
 * @dete 2019/4/20
 * @description 自定義DataSource
 *
 */
@Configuration
public class DruidDBConfig {

    private Logger logger = LoggerFactory.getLogger(DruidDBConfig.class);

    @Value("${spring.datasource.url}")
    private String dbUrl;

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.driver-class-name}")
    private String driverClassName;

   /* @Value("${spring.datasource.initialSize}")
    private int initialSize;

    @Value("${spring.datasource.minIdle}")
    private int minIdle;

    @Value("${spring.datasource.maxActive}")
    private int maxActive;

    @Value("${spring.datasource.maxWait}")
    private int maxWait;

    @Value("${spring.datasource.timeBetweenEvictionRunsMillis}")
    private int timeBetweenEvictionRunsMillis;

    @Value("${spring.datasource.minEvictableIdleTimeMillis}")
    private int minEvictableIdleTimeMillis;

    @Value("${spring.datasource.validationQuery}")
    private String validationQuery;

    @Value("${spring.datasource.testWhileIdle}")
    private boolean testWhileIdle;

    @Value("${spring.datasource.testOnBorrow}")
    private boolean testOnBorrow;

    @Value("${spring.datasource.testOnReturn}")
    private boolean testOnReturn;

    @Value("${spring.datasource.poolPreparedStatements}")
    private boolean poolPreparedStatements;

    @Value("${spring.datasource.maxPoolPreparedStatementPerConnectionSize}")
    private int maxPoolPreparedStatementPerConnectionSize;

    @Value("${spring.datasource.filters}")
    private String filters;

    @Value("{spring.datasource.connectionProperties}")
    private String connectionProperties;*/

    @Bean
    @Primary  // 被注入的優先級最高
    public DataSource dataSource() {
        DruidDataSource dataSource = new DruidDataSource();
        logger.info("-------->dataSource[url="+dbUrl+" ,username="+username+"]");
        dataSource.setUrl(dbUrl);
        dataSource.setUsername(username);
        dataSource.setPassword(password);
        dataSource.setDriverClassName(driverClassName);

        /*  //configuration
        datasource.setInitialSize(initialSize);
        datasource.setMinIdle(minIdle);
        datasource.setMaxActive(maxActive);
        datasource.setMaxWait(maxWait);
        datasource.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
        datasource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
        datasource.setValidationQuery(validationQuery);
        datasource.setTestWhileIdle(testWhileIdle);
        datasource.setTestOnBorrow(testOnBorrow);
        datasource.setTestOnReturn(testOnReturn);
        datasource.setPoolPreparedStatements(poolPreparedStatements);
        datasource.setMaxPoolPreparedStatementPerConnectionSize(maxPoolPreparedStatementPerConnectionSize);
        try {
            datasource.setFilters(filters);
        } catch (SQLException e) {
            logger.error("druid configuration initialization filter", e);
        }
        datasource.setConnectionProperties(connectionProperties);*/

        return dataSource;
    }

    @Bean
    public ServletRegistrationBean druidServletRegistrationBean() {
        ServletRegistrationBean servletRegistrationBean = new ServletRegistrationBean();
        servletRegistrationBean.setServlet(new StatViewServlet());
        servletRegistrationBean.addUrlMappings("/druid/*");
        return servletRegistrationBean;
    }

    /**
     * 註冊DruidFilter攔截
     *
     * @return
     */
    @Bean
    public FilterRegistrationBean duridFilterRegistrationBean() {
        FilterRegistrationBean filterRegistrationBean = new FilterRegistrationBean();
        filterRegistrationBean.setFilter(new WebStatFilter());
        Map<String, String> initParams = new HashMap<String, String>();
        //設置忽略請求
        initParams.put("exclusions", "*.js,*.gif,*.jpg,*.bmp,*.png,*.css,*.ico,/druid/*");
        filterRegistrationBean.setInitParameters(initParams);
        filterRegistrationBean.addUrlPatterns("/*");
        return filterRegistrationBean;
    }
}

View Code

四、編寫batch配置類

在配置類中，註冊Spring Batch的各個組成部分便可，其中部分說明已在代碼中註釋.

/**
 *
 * @author jian
 * @date 2019/4/28
 * @description spring batch cvs文件批處理配置須要注入Spring Batch如下組成部分
 * spring batch組成：
 * 1）JobRepository 註冊job的容器
 * 2）JonLauncher 用來啓動job的接口
 * 3）Job 實際執行的任務，包含一個或多個Step
 * 4）Step Step步驟包括ItemReader、ItemProcessor和ItemWrite
 * 5）ItemReader 讀取數據的接口
 * 6）ItemProcessor 處理數據的接口
 * 7）ItemWrite 輸出數據的接口
 *
 *
 */
@Configuration
@EnableBatchProcessing // 開啓批處理的支持
@Import(DruidDBConfig.class) // 注入datasource
public class CsvBatchConfig {
    private Logger logger = LoggerFactory.getLogger(CsvBatchConfig.class);


    /**
     * ItemReader定義：讀取文件數據+entirty映射
     * @return
     */
    @Bean
    public ItemReader<Person> reader(){
        // 使用FlatFileItemReader去讀cvs文件，一行即一條數據
        FlatFileItemReader<Person> reader = new FlatFileItemReader<>();
        // 設置文件處在路徑
        reader.setResource(new ClassPathResource("person.csv"));
        // entity與csv數據作映射
        reader.setLineMapper(new DefaultLineMapper<Person>() {
            {
                setLineTokenizer(new DelimitedLineTokenizer() {
                    {
                        setNames(new String[]{"id", "name", "age", "gender"});
                    }
                });
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {
                    {
                        setTargetType(Person.class);
                    }
                });
            }
        });
        return reader;
    }


    /**
     * 註冊ItemProcessor: 處理數據+校驗數據
     * @return
     */
    @Bean
    public ItemProcessor<Person, Person> processor(){
        CvsItemProcessor cvsItemProcessor = new CvsItemProcessor();
        // 設置校驗器
        cvsItemProcessor.setValidator(csvBeanValidator());
        return cvsItemProcessor;
    }

    /**
     * 註冊校驗器
     * @return
     */
    @Bean
    public CsvBeanValidator csvBeanValidator(){
        return new CsvBeanValidator<Person>();
    }

    /**
     * ItemWriter定義：指定datasource，設置批量插入sql語句，寫入數據庫
     * @param dataSource
     * @return
     */
    @Bean
    public ItemWriter<Person> writer(DataSource dataSource){
        // 使用jdbcBcatchItemWrite寫數據到數據庫中
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        // 設置有參數的sql語句
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
        String sql = "insert into person values(:id,:name,:age,:gender)";
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    /**
     * JobRepository定義：設置數據庫，註冊Job容器
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public JobRepository cvsJobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        jobRepositoryFactoryBean.setTransactionManager(transactionManager);
        jobRepositoryFactoryBean.setDataSource(dataSource);
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * jobLauncher定義：
     * @param dataSource
     * @param transactionManager
     * @return
     * @throws Exception
     */
    @Bean
    public SimpleJobLauncher csvJobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        // 設置jobRepository
        jobLauncher.setJobRepository(cvsJobRepository(dataSource, transactionManager));
        return jobLauncher;
    }

    /**
     * 定義job
     * @param jobs
     * @param step
     * @return
     */
    @Bean
    public Job importJob(JobBuilderFactory jobs, Step step){
        return jobs.get("importCsvJob")
                .incrementer(new RunIdIncrementer())
                .flow(step)
                .end()
                .listener(csvJobListener())
                .build();
    }

    /**
     * 註冊job監聽器
     * @return
     */
    @Bean
    public CsvJobListener csvJobListener(){
        return new CsvJobListener();
    }


    /**
     * step定義：步驟包括ItemReader->ItemProcessor->ItemWriter 即讀取數據->處理校驗數據->寫入數據
     * @param stepBuilderFactory
     * @param reader
     * @param writer
     * @param processor
     * @return
     */
    @Bean
    public Step step(StepBuilderFactory stepBuilderFactory, ItemReader<Person> reader,
                     ItemWriter<Person> writer, ItemProcessor<Person, Person> processor){
        return stepBuilderFactory
                .get("step")
                .<Person, Person>chunk(65000) // Chunk的機制(即每次讀取一條數據，再處理一條數據，累積到必定數量後再一次性交給writer進行寫入操做)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();

    }
}

View Code

五、定義處理器

只須要實現ItemProcessor接口，重寫process方法，輸入的參數是從ItemReader讀取到的數據，返回的數據給ItemWriter

/**
 * @author jian
 * @date 2019/4/28
 * @description
 * CSV文件數據處理及校驗
 * 只須要實現ItemProcessor接口，重寫process方法，輸入的參數是從ItemReader讀取到的數據，返回的數據給ItemWriter
 */
public class CvsItemProcessor extends ValidatingItemProcessor<Person> {
    private Logger logger = LoggerFactory.getLogger(CvsItemProcessor.class);

    @Override
    public Person process(Person item) throws ValidationException {
        // 執行super.process()才能調用自定義的校驗器
        logger.info("processor start validating...");
        super.process(item);

        // 數據處理，好比將中文性別設置爲M/F
        if ("男".equals(item.getGender())) {
            item.setGender("M");
        } else {
            item.setGender("F");
        }
        logger.info("processor end validating...");
        return item;
    }
}

六、定義校驗器

定義校驗器：使用JSR-303(hibernate-validator)註解，來校驗ItemReader讀取到的數據是否知足要求。如不知足則不會進行接下來的批處理任務。

/**
 *
 * @author jian
 * @date 2019/4/28
 * @param <T>
 * @description 定義校驗器：使用JSR-303(hibernate-validator)註解，來校驗ItemReader讀取到的數據是否知足要求。
 */

public class CsvBeanValidator<T> implements Validator<T>, InitializingBean {

    private javax.validation.Validator validator;


    /**
     * 進行JSR-303的Validator的初始化
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
        validator = validatorFactory.usingContext().getValidator();
    }

    /**
     * 使用validator方法檢驗數據
     * @param value
     * @throws ValidationException
     */
    @Override
    public void validate(T value) throws ValidationException {
        Set<ConstraintViolation<T>> constraintViolations = validator.validate(value);
        if (constraintViolations.size() > 0) {
            StringBuilder message = new StringBuilder();
            for (ConstraintViolation<T> constraintViolation: constraintViolations) {
                message.append(constraintViolation.getMessage() + "\n");
            }
            throw new ValidationException(message.toString());
        }
    }
}

View Code

七、定義監聽器：

監聽Job執行狀況，則定義一個類實現JobExecutorListener，並定義Job的Bean上綁定該監聽器

/**
 * @author jian
 * @date 2019/4/28
 * @description
 * 監聽Job執行狀況，則定義一個類實現JobExecutorListener，並定義Job的Bean上綁定該監聽器
 */
public class CsvJobListener implements JobExecutionListener {

    private Logger logger = LoggerFactory.getLogger(CsvJobListener.class);
    private long startTime;
    private long endTime;

    @Override
    public void beforeJob(JobExecution jobExecution) {
        startTime = System.currentTimeMillis();
        logger.info("job process start...");
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        endTime = System.currentTimeMillis();
        logger.info("job process end...");
        logger.info("elapsed time: " + (endTime - startTime) + "ms");
    }
}

3、測試

一、person.csv文件

csv文件時以逗號爲分隔的數據表示字段，回車表示一行（條）數據記錄

1,Zhangsan,21,男
2,Lisi,22,女
3,Wangwu,23,男
4,Zhaoliu,24,男
5,Zhouqi,25,女

放在resources下，在ItemReader中讀取的該路徑便可

二、person實體

person.csv中的字段與之對應，並在該實體中能夠添加校驗註解，如@Size表示該字段的長度範圍，若是超過規定。則會被校驗檢測到，批處理將不會進行！

public class Person implements Serializable {
    private final long serialVersionUID = 1L;

    private String id;
    @Size(min = 2, max = 8)
    private String name;
    private int age;
    private String gender;

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getGender() {
        return gender;
    }

    public void setGender(String gender) {
        this.gender = gender;
    }

    @Override
    public String toString() {
        return "Person{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", age=" + age +
                ", gender='" + gender + '\'' +
                '}';
    }
}

View Code

三、數據表

CREATE TABLE `person` (
  `id` int(11) NOT NULL,
  `name` varchar(10) DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  `gender` varchar(2) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1

一開始表是沒有數據的

四、測試類

須要注入發佈器，與job任務。同時可使用後置參數靈活處理，最後調用JobLauncher.run方法執行批處理任務

@RunWith(SpringRunner.class)
@SpringBootTest
public class BatchTest {

    @Autowired
    SimpleJobLauncher jobLauncher;

    @Autowired
    Job importJob;

    @Test
    public void test() throws Exception{
        // 後置參數：使用JobParameters中綁定參數
        JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis())
                .toJobParameters();
        jobLauncher.run(importJob, jobParameters);
    }
}

五、測試結果

....
2019-05-09 15:23:39.576  INFO 18296 --- [           main] com.lijian.test.BatchTest                : Started BatchTest in 6.214 seconds (JVM running for 7.185)
2019-05-09 15:23:39.939  INFO 18296 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] launched with the following parameters: [{time=1557386619763}]
2019-05-09 15:23:39.982  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : job process start...
2019-05-09 15:23:40.048  INFO 18296 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step]
2019-05-09 15:23:40.214  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:23:40.282  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:23:40.284  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
2019-05-09 15:23:40.525  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : job process end...
2019-05-09 15:23:40.526  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : elapsed time: 543ms
2019-05-09 15:23:40.548  INFO 18296 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] completed with the following parameters: [{time=1557386619763}] and the following status: [COMPLETED]
2019-05-09 15:23:40.564  INFO 18296 --- [       Thread-5] com.alibaba.druid.pool.DruidDataSource   : {dataSource-1} closed

查看錶中數據： select * from person;

若繼續插入數據，而且測試校驗器是否生效，則將person.csv更改成以下內容：

6,springbatch,24,男
7,springboot,23,女

因爲實體類中JSR校驗註解對name長度範圍進行了檢驗，即添加了 @Size(min=2, max=8) 的註解。故會報錯顯示校驗不經過，批處理將不會進行。

...
Started BatchTest in 5.494 seconds (JVM running for 6.41)
2019-05-09 15:30:02.147  INFO 20368 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] launched with the following parameters: [{time=1557387001499}]
2019-05-09 15:30:02.247  INFO 20368 --- [           main] com.lijian.config.batch.CsvJobListener   : job process start...
2019-05-09 15:30:02.503  INFO 20368 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step]
2019-05-09 15:30:02.683  INFO 20368 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
2019-05-09 15:30:02.761 ERROR 20368 --- [           main] o.s.batch.core.step.AbstractStep         : Encountered an error executing step step in job importCsvJob

org.springframework.batch.item.validator.ValidationException: size must be between 2 and 8
...

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。