spring batch + spring boot 配置

淩亂°似流年 2022-08-10 00:51 332阅读 0赞

  1. spring batch 批处理配置

    import java.io.IOException;

    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    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.configuration.annotation.StepScope;
    import org.springframework.batch.core.launch.support.RunIdIncrementer;
    import org.springframework.batch.core.repository.JobRepository;
    import org.springframework.batch.item.ItemProcessor;
    import org.springframework.batch.item.ItemReader;
    import org.springframework.batch.item.ItemWriter;
    import org.springframework.batch.item.ParseException;
    import org.springframework.batch.item.file.FlatFileItemReader;
    import org.springframework.batch.item.file.LineMapper;
    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.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.core.io.ClassPathResource;
    import org.springframework.core.task.SimpleAsyncTaskExecutor;
    import org.springframework.transaction.PlatformTransactionManager;

    /**

    • spring batch 配置
    • @author
      */
      @Configuration
      @EnableBatchProcessing
      public class BlackListBatchConfiguration {
  1.     private static final Logger logger = LoggerFactory.getLogger(BlackListBatchConfiguration.class);
  2.     /**
  3.      * 读取外部文件方法
  4.      * @return
  5.      * @throws IOException 
  6.      */
  7.     @Bean
  8.     @StepScope
  9.     public ItemReader<BlackListDO> reader(@Value("#{jobParameters[inputFileBlack]}") String inputFile) throws IOException { 
  10.         logger.info("inputFile:"+new ClassPathResource(inputFile).getURL().getPath());
  11.         if(inputFile == null){
  12.             logger.error("The blacklist reader file is null");
  13.             return null;
  14.         }
  15.         FlatFileItemReader<BlackListDO> reader = new FlatFileItemReader<BlackListDO>();
  17.         reader.setResource(new ClassPathResource(inputFile));
  19.         reader.setLineMapper(lineMapper());
  21.         reader.setLinesToSkip(1);
  23.         reader.open(JobCompletionNotificationListener.jobExecution.getExecutionContext());
  25.         return reader;
  26.     }
  29.     /**
  30.      * 读取文本行映射POJO
  31.      * @return
  32.      */
  33.     @Bean
  34.     @StepScope
  35.     public LineMapper<BlackListDO> lineMapper() {
  36.         DefaultLineMapper<BlackListDO> lineMapper = new DefaultLineMapper<BlackListDO>();
  37.         DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
  38.         lineTokenizer.setDelimiter(",");
  39.         lineTokenizer.setStrict(false);
  40.         lineTokenizer.setNames(new String[] { "type","value","fraudType"});
  42.         BeanWrapperFieldSetMapper<BlackListDO> fieldSetMapper = new BeanWrapperFieldSetMapper<BlackListDO>();
  43.         fieldSetMapper.setTargetType(BlackListDO.class);
  44.         lineMapper.setLineTokenizer(lineTokenizer);
  45.         lineMapper.setFieldSetMapper(new BlackListFieldSetMapper());
  46.         return lineMapper;
  47.     }
  49.     /**
  50.      * 处理过程
  51.      * @return
  52.      */
  53.     @Bean
  54.     @StepScope
  55.     public ItemProcessor<BlackListDO, BlackListDO> processor(@Value("#{jobParameters[inputFileBlack]}") String inputFile) {
  56.         return new BlackListDOItemProcessor(inputFile);
  57.     }
  59.     /**
  60.      * 写出内容
  61.      * @return
  62.      */
  63.     @Bean
  64.     @StepScope
  65.     public ItemWriter<BlackListDO> writer() {
  66.         return new BlackListItemWriter();
  67.     }
  69.     /**
  70.      * 构建job
  71.      * @param jobs
  72.      * @param s1 
  73.      * @param listener
  74.      * @return
  75.      */
  76.     @Bean
  77.     public Job importFileJob(JobBuilderFactory jobs, Step step1,JobExecutionListener listener,JobRepository jobRepository) {
  78.         return jobs.get("importFileJob")
  79.                 .incrementer(new RunIdIncrementer())
  80.                 .repository(jobRepository)
  81.                 .listener(listener)
  82.                 .flow(step1)
  83.                 .end()
  84.                 .build();
  85.     }
  87.     /**
  88.      * 声明step
  89.      * @param stepBuilderFactory 
  90.      * @param reader 
  91.      * @param writer
  92.      * @param processor
  93.      * @return
  94.      */
  95.     @Bean
  96.     public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<BlackListDO> reader,
  97.             ItemWriter<BlackListDO> writer, ItemProcessor<BlackListDO, BlackListDO> processor,PlatformTransactionManager transactionManager) {
  98.         logger.error("step1");
  99.         return stepBuilderFactory.get("step1")
  100.                 .<BlackListDO, BlackListDO> chunk(500)
  101.                 .reader(reader)
  102.                 .processor(processor)
  103.                 .writer(writer)
  104.                 .faultTolerant()
  105.                 .retry(Exception.class)   // 重试
  106.                 .noRetry(ParseException.class)
  107.                 .retryLimit(1)           //每条记录重试一次
  108.                 .listener(new RetryFailuireItemListener())  
  109.                 .skip(Exception.class)
  110.                 .skipLimit(500)         //一共允许跳过200次异常
  111.                 .taskExecutor(new SimpleAsyncTaskExecutor()) //设置并发方式执行
  112.                 .throttleLimit(10)        //并发任务数为 10,默认为4
  113.                 .transactionManager(transactionManager)
  114.                 .build();
  115.     }
  117. }

BlackListDOItemProcessor 处理类

  1. import java.sql.Timestamp;
  2. import java.util.Date;
  3. import java.util.UUID;
  5. import org.springframework.batch.item.ItemProcessor;
  7. import com.BlackListDO;
  9. /**
  10.  * @author zhengyong
  11.  *
  12.  */
  14. public class BlackListDOItemProcessor implements ItemProcessor<BlackListDO, BlackListDO> {
  16.     public String inputFile;
  18.     public BlackListDOItemProcessor() {
  19.     }
  21.     public BlackListDOItemProcessor(String inputFile) {
  22.         this.inputFile = inputFile;
  23.     }
  24.     // 数据处理
  25.     public BlackListDO process(BlackListDO blackListDO) throws Exception {
  26.         blackListDO.setDeleteFlag(0);
  27.         blackListDO.setUuid(UUID.randomUUID().toString().replaceAll("-", ""));
  28.         return blackListDO;
  29.     }
  31. }

BlackListItemWriter 写入数据库类

  1. import org.springframework.batch.item.ItemWriter;
  2. import org.springframework.beans.factory.annotation.Autowired;
  4. import com.BlackListDO;
  6. public class BlackListItemWriter implements ItemWriter<BlackListDO> {
  8.     @Override
  9.     public void write(List<? extends BlackListDO> blackList) throws Exception {
  10.             // 插入数据库操作 
  11.     }
  13. }

JobCompletionNotificationListener 监听任务

  1. import org.springframework.batch.core.BatchStatus;
  2. import org.springframework.batch.core.JobExecution;
  3. import org.springframework.batch.core.listener.JobExecutionListenerSupport;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Component;
  7. @Component
  8. public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
  10.     @Override
  11.     public void afterJob(JobExecution jobExecution) {
  13.     }
  15.     @Override
  16.     public void beforeJob(JobExecution jobExecution) {
  17.         super.beforeJob(jobExecution);
  18.     }
  19. }

RetryFailuireItemListener 重试监听

  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  3. import org.springframework.retry.RetryCallback;
  4. import org.springframework.retry.RetryContext;
  5. import org.springframework.retry.RetryListener;
  7. public class RetryFailuireItemListener implements RetryListener{
  9.     private static final Logger logger = LoggerFactory.getLogger(RetryFailuireItemListener.class);
  11.     @Override
  12.     public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
  13.         return true;
  14.     }
  16.     @Override
  17.     public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
  18.             Throwable throwable) {
  19.     }
  21.     @Override
  22.     public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
  23.             Throwable throwable) {
  24.         logger.error("【重试异常】："+throwable.getMessage());
  25.     }
  27. }

  1. spring boot 配置

    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.boot.builder.SpringApplicationBuilder;
    import org.springframework.boot.context.web.SpringBootServletInitializer;
    import org.springframework.context.annotation.ComponentScan;

    @ComponentScan(“com.syncclient”)
    @SpringBootApplication
    public class SpringBootJspApplication extends SpringBootServletInitializer{

    1.  /**
    2.   * 500一批
    3.   * oracle : 单条插入基本每分钟2.5W条（50W，19.5min） ,批量插入基本每分钟10W条(50W，4.7mim)
    4.   * mysql  : 单条插入基本每分钟2.5W条（50W，11.4min） ,批量插入基本每分钟40W条(50W，1.3min)
    5.   */
    6.  @Override
    7.  protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
    8.      return application.sources(SpringBootJspApplication.class);
    9.  }
    11.  public static void main(String[] args) throws Exception {
    12.      SpringApplication.run(SpringBootJspApplication.class,new String[]{"appStart=true"});
    13.  }

    }

  2. 配置文件

    mysql config

    spring.boot.database = mysql

    //127.0.0.1:3306/spring_batch?useUnicode=true&characterEncoding=utf8" class="reference-link">spring.datasource.url = jdbc:mysql://127.0.0.1:3306/spring_batch?useUnicode=true&characterEncoding=utf8

    spring.datasource.username = admin

    spring.datasource.password = 123456

    spring.datasource.driverClassName = com.mysql.jdbc.Driver

    spring.batch.schema = classpath:/org/springframework/batch/core/schema-mysql.sql

    spring.batch.drop = classpath:/org/springframework/batch/core/schema-drop-mysql.sql

    oracle config

    spring.boot.database = oracle
    spring.datasource.url = jdbc:oracle:thin:@127.0.0.1:1521:spring_batch
    spring.datasource.username = admin
    spring.datasource.password = 123456
    spring.datasource.driverClassName = oracle.jdbc.driver.OracleDriver
    spring.batch.schema = classpath:org/springframework/batch/core/schema-oracle10g.sql
    spring.batch.drop = classpath:org/springframework/batch/core/schema-drop-oracle10g.sql

    batch config

    spring.batch.job.names = importFileJob
    spring.batch.job.enabled = true
    spring.batch.initializer.enabled=true

  3. mybatis 配置

    import javax.sql.DataSource;

    import org.apache.ibatis.session.SqlSessionFactory;
    import org.mybatis.spring.SqlSessionFactoryBean;
    import org.mybatis.spring.SqlSessionTemplate;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
    import org.springframework.jdbc.datasource.DataSourceTransactionManager;
    import org.springframework.transaction.PlatformTransactionManager;

    import com.alibaba.druid.pool.DruidDataSource;

    import com.BlackListDao;
    import com.ProperitesUtil;

    /**

    • mybatis配置
      /
      @Configuration
      public class MyBatisConfiguration {

      private static final Logger logger = LoggerFactory.getLogger(MyBatisConfiguration.class);

      @Autowired
      SqlSessionFactory sqlSessionFactory;
      @Autowired
      SqlSessionTemplate sessionTemplate;

      @Bean
      public SqlSessionTemplate sqlSessionTemplate() {

      1.  return new SqlSessionTemplate(sqlSessionFactory);

      }

      @Bean
      public BlackListDao blackListMapper() {

      1.  return sessionTemplate.getMapper(BlackListDao.class);

      }

      @Bean
      @Primary
      @ConfigurationProperties(prefix = “spring.datasource”)
      public DataSource dataSource() {

      1.  logger.debug("初始化dataSource");
      2.  return new DruidDataSource();

      }

      @Bean
      public PlatformTransactionManager transactionManager(DataSource dataSource) {

      1.  PlatformTransactionManager transactionManager = new DataSourceTransactionManager(dataSource);
      2.  return transactionManager;

      }

      @Bean
      public SqlSessionFactory sqlSessionFactory() throws Exception {

      1.  SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
      2.  sqlSessionFactoryBean.setDataSource(dataSource());
      4.  PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
      6.  // 获取数据库类型
      7.  String dataBaseType = ProperitesUtil.getPropertyValue("spring.boot.database") == null ? "mysql"
      8.          : ProperitesUtil.getPropertyValue("spring.boot.database");
      10.  String directory = "classpath:/mapper/" + dataBaseType.trim().toLowerCase() + "/*.xml";
      11.  sqlSessionFactoryBean.setMapperLocations(resolver.getResources(directory));
      13.  return sqlSessionFactoryBean.getObject();

      }

      }

  1. pom.xml文件


    4.0.0

    com.spring.batch
    batch-test
    0.0.1-SNAPSHOT
    jar

    batch-test
    http://maven.apache.org


    org.springframework.boot
    spring-boot-starter-parent
    1.2.5.RELEASE




    org.springframework.boot
    spring-boot-starter-web


    org.springframework.boot
    spring-boot-starter-batch


    org.springframework.batch
    spring-batch-core
    3.0.4.RELEASE


    org.springframework.boot
    spring-boot-starter-tomcat
    provided


    org.apache.tomcat.embed
    tomcat-embed-jasper
    provided


    javax.servlet
    jstl


    org.springframework.boot
    spring-boot-starter-test
    test

发表评论

表情：
评论列表 （有 0 条评论，332人围观）

还没有评论，来说两句吧...

相关阅读

    相关 spring batch

            springBatch一个轻量级，全面的批处理框架，旨在开发强大的批处理应用程序，对于企业系统的日常运营至关重要。提供可重复使用的功能，这些功能在处理大量记录

    野性酷女野性酷女/ 0 赞/ 244 阅读

    相关 Spring Boot整合Spring Batch

    引言 > 　　Spring Batch是处理大量数据操作的一个框架，主要用来读取大量数据，然后进行一定的处理后输出指定的形式。比如我们可以将csv文件中的数据（数据量几百

    快来打我*快来打我*/ 0 赞/ 459 阅读