记一次Spring Batch完整入门实践

╰+攻爆jí腚メ 2023-06-16 07:55 22阅读 0赞

# 1.Spring Batch 结构 #

网上有很多Spring Batch结构和原理的讲解，我就不详细阐述了，我这里只讲一下Spring Batch的一个基本层级结构。

首先，Spring Batch运行的基本单位是一个Job，一个Job就做一件批处理的事情。一个Job包含很多Step，step就是每个job要执行的单个步骤。

如下图所示，Step里面，会有Tasklet，Tasklet是一个任务单元，它是属于可以重复利用的东西。然后是Chunk，chunk就是数据块，你需要定义多大的数据量是一个chunk。

Chunk里面就是不断循环的一个流程，读数据，处理数据，然后写数据。Spring Batch会不断的循环这个流程，直到批处理数据完成。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMxOTQyNQ_size_16_color_FFFFFF_t_70][]

# 2.构建Spring Batch #

首先，我们需要一个全局的Configuration来配置所有的Job和一些全局配置。

代码如下：

@Configuration
    @EnableAutoConfiguration
    @EnableBatchProcessing(modular = true)
    public class SpringBatchConfiguration {
        @Bean
        public ApplicationContextFactory firstJobContext() {
            return new GenericApplicationContextFactory(FirstJobConfiguration.class);
        }
        
        @Bean
        public ApplicationContextFactory secondJobContext() {
            return new GenericApplicationContextFactory(SecondJobConfiguration.class);
        }
    
    }

@EnableBatchProcessing是打开Batch。如果要实现多Job的情况，需要把**EnableBatchProcessing**注解的**modular**设置为**true**，让每个Job使用自己的ApplicationConext。

比如上面代码的就创建了两个Job。

# 3.构建Job #

首先我们需要一个关于这个Job的Configuration，它将在SpringBatchConfigration里面被加载。

@Configuration
    @EnableAutoConfiguration
    @EnableBatchProcessing(modular = true)
    public class SpringBatchConfiguration {
        @Bean
        public ApplicationContextFactory messageMigrationJobContext() {
            return new GenericApplicationContextFactory(MessageMigrationJobConfiguration.class);
        }
    }

下面的关于构建Job的代码都将写在这个MessageMigrationJobConfiguration里面。

public class MessageMigrationJobConfiguration {
    }

我们先定义一个Job的Bean。

@Autowired
    private JobBuilderFactory jobBuilderFactory;
    
    @Bean
    public Job messageMigrationJob(@Qualifier("messageMigrationStep") Step messageMigrationStep) {
        return jobBuilderFactory.get("messageMigrationJob")
                .start(messageMigrationStep)
                .build();
    }

jobBuilderFactory是注入进来的，get里面的就是job的名字。这个job只有一个step。

# 4.Step #

接下来就是创建Step。

@Autowired
    private StepBuilderFactory stepBuilderFactory;
    
    @Bean
    public Step messageMigrationStep(@Qualifier("jsonMessageReader") FlatFileItemReader<Message> jsonMessageReader,
                                     @Qualifier("messageItemWriter") JpaItemWriter<Message> messageItemWriter,
                                     @Qualifier("errorWriter") Writer errorWriter) {
        return stepBuilderFactory.get("messageMigrationStep")
                .<Message, Message>chunk(CHUNK_SIZE)
                .reader(jsonMessageReader).faultTolerant().skip(JsonParseException.class).skipLimit(SKIP_LIMIT)
                .listener(new MessageItemReadListener(errorWriter))
                .writer(messageItemWriter).faultTolerant().skip(Exception.class).skipLimit(SKIP_LIMIT)
                .listener(new MessageWriteListener())
                .build();
    }

stepBuilderFactory是注入进来的，然后get里面是Step的名字。  
我们的Step中可以构建很多东西，比如reader，processer，writer，listener等等。

下面我们就逐个来看看step里面的这些东西是如何使用的。

# 5.Chunk #

Spring batch在配置Step时采用的是基于Chunk的机制，即每次读取一条数据，再处理一条数据，累积到一定数量后再一次性交给writer进行写入操作。这样可以最大化的优化写入效率，整个事务也是基于Chunk来进行。

比如我们定义chunk size是50，那就意味着，spring batch处理了50条数据后，再统一向数据库写入。  
这里有个很重要的点，chunk前面需要定义数据输入类型和输出类型，由于我们输入是Message，输出也是Message，所以两个都直接写Message了。  
如果不定义这个类型，会报错。

.<Message, Message>chunk(CHUNK_SIZE)

# 6.Reader #

Reader顾名思义就是从数据源读取数据。  
Spring Batch给我们提供了很多好用实用的reader，基本能满足我们所有需求。比如FlatFileItemReader，JdbcCursorItemReader，JpaPagingItemReader等。也可以自己实现Reader。

本例子里面，数据源是文本文件，所以我们就使用FlatFileItemReader。FlatFileItemReader是从文件里面一行一行的读取数据。  
首先需要设置文件路径，也就是设置resource。  
因为我们需要把一行文本映射为Message类，所以我们需要自己设置并实现LineMapper。

@Bean
    public FlatFileItemReader<Message> jsonMessageReader() {
        FlatFileItemReader<Message> reader = new FlatFileItemReader<>();
        reader.setResource(new FileSystemResource(new File(MESSAGE_FILE)));
        reader.setLineMapper(new MessageLineMapper());
        return reader;
    }

# 7.Line Mapper #

LineMapper的输入就是获取一行文本，和行号，然后转换成Message。  
在本例子里面，一行文本就是一个json对象，所以我们使用JsonParser来转换成Message。

public class MessageLineMapper implements LineMapper<Message> {
        private MappingJsonFactory factory = new MappingJsonFactory();
    
        @Override
        public Message mapLine(String line, int lineNumber) throws Exception {   
            JsonParser parser = factory.createParser(line);
            Map<String, Object> map = (Map) parser.readValueAs(Map.class);
            Message message = new Message();
            ... // 转换逻辑
            return message;
        }
    }

# 8.Processor #

由于本例子里面，数据是一行文本，通过reader变成Message的类，然后writer直接把Message写入MySQL。所以我们的例子里面就不需要Processor，关于如何写Processor其实和reader/writer是一样的道理。  
从它的接口可以看出，需要定义输入和输出的类型，把输入I通过某些逻辑处理之后，返回输出O。

public interface ItemProcessor<I, O> {
        O process(I item) throws Exception;
    }

# 9.Writer #

Writer顾名思义就是把数据写入到目标数据源里面。  
Spring Batch同样给我们提供很多好用实用的writer。比如JpaItemWriter，FlatFileItemWriter，HibernateItemWriter，JdbcBatchItemWriter等。同样也可以自定义。

本例子里面，使用的是JpaItemWriter，可以直接把Message对象写到数据库里面。但是需要设置一个EntityManagerFactory，可以注入进来。

@Autowired
    private EntityManagerFactory entityManager;
    
    @Bean
    public JpaItemWriter<Message> messageItemWriter() {
        JpaItemWriter<Message> writer = new JpaItemWriter<>();
        writer.setEntityManagerFactory(entityManager);
        return writer;
    }

另外，你需要配置数据库的连接等东西。由于我使用的spring，所以直接在Application.properties里面配置如下：

spring.datasource.url=jdbc:mysql://database
    spring.datasource.username=username
    spring.datasource.password=password
    spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
    spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect
    spring.jpa.show-sql=true
    spring.jpa.properties.jadira.usertype.autoRegisterUserTypes=true
    spring.jackson.serialization.write-dates-as-timestamps=false
    spring.batch.initialize-schema=ALWAYS
    spring.jpa.hibernate.ddl-auto=update

spring.datasource相关的设置都是在配置数据库的连接。  
spring.batch.initialize-schema=always表示让spring batch在数据库里面创建默认的数据表。  
spring.jpa.show-sql=true表示在控制台输出hibernate读写数据库时候的SQL。  
spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect是在指定MySQL的方言。

# 10.Listener #

Spring Batch同样实现了非常完善全面的listener，listener很好理解，就是用来监听每个步骤的结果。比如可以有监听step的，有监听job的，有监听reader的，有监听writer的。没有你找不到的listener，只有你想不到的listener。

在本例子里面，我只关心，read的时候有没有出错，和write的时候有没有出错，所以，我只实现了ReadListener和WriteListener。  
在read出错的时候，把错误结果写入一个单独的error列表文件中。

public class MessageItemReadListener implements ItemReadListener<Message> {
        private Writer errorWriter;
    
        public MessageItemReadListener(Writer errorWriter) {
            this.errorWriter = errorWriter;
        }
    
        @Override
        public void beforeRead() {
        }
    
        @Override
        public void afterRead(Message item) {
        }
    
        @Override
        public void onReadError(Exception ex) {
             errorWriter.write(format("%s%n", ex.getMessage()));
        }
    }

在write出错的时候，也做同样的事情，把出错的原因写入单独的日志中。

public class MessageWriteListener implements ItemWriteListener<Message> {
    
        @Autowired
        private Writer errorWriter;
    
        @Override
        public void beforeWrite(List<? extends Message> items) {
        }
    
        @Override
        public void afterWrite(List<? extends Message> items) {
        }
    
        @Override
        public void onWriteError(Exception exception, List<? extends Message> items) {
            errorWriter.write(format("%s%n", exception.getMessage()));
            for (Message message : items) {
                errorWriter.write(format("Failed writing message id: %s", message.getObjectId()));
            }
        }
    }

前面有说chuck机制，所以write的listener传入参数是一个List，因为它是累积到一定的数量才一起写入。

# 11.Skip #

Spring Batch提供了skip的机制，也就是说，如果出错了，可以跳过。如果你不设置skip，那么一条数据出错了，整个job都会挂掉。  
设置skip的时候一定要设置什么Exception才需要跳过，并且跳过多少条数据。如果失败的数据超过你设置的skip limit，那么job就会失败。  
你可以分别给reader和writer等设置skip机制。

writer(messageItemWriter).faultTolerant().skip(Exception.class).skipLimit(SKIP_LIMIT)

# 12.Retry #

这个和Skip是一样的原理，就是失败之后可以重试，你同样需要设置重试的次数。  
同样可以分别给reader，writer等设置retry机制。

如果同时设置了retry和skip，会先重试所有次数，然后再开始skip。比如retry是10次，skip是20，会先重试10次之后，再开始算第一次skip。

# 13.运行Job #

所有东西都准备好以后，就是如何运行了。  
运行就是在main方法里面用JobLauncher去运行你制定的job。

下面是我写的main方法，main方法的第一个参数是job的名字，这样我们就可以通过不同的job名字跑不同的job了。

首先我们通过运行起来的Spring application得到jobRegistry，然后通过job的名字找到对应的job。

接着，我们就可以用jobLauncher去运行这个job了，运行的时候会传一些参数，比如你job里面需要的文件路径或者文件日期等，就可以通过这个jobParameters传进去。如果没有参数，可以默认传当前时间进去。

public static void main(String[] args) {
        String jobName = args[0];
    
        try {
            ConfigurableApplicationContext context = SpringApplication.run(ZuociBatchApplication.class, args);
            JobRegistry jobRegistry = context.getBean(JobRegistry.class);
            Job job = jobRegistry.getJob(jobName);
            JobLauncher jobLauncher = context.getBean(JobLauncher.class);
            JobExecution jobExecution = jobLauncher.run(job, createJobParams());
            if (!jobExecution.getExitStatus().equals(ExitStatus.COMPLETED)) {
                throw new RuntimeException(format("%s Job execution failed.", jobName));
            }
        } catch (Exception e) {
            throw new RuntimeException(format("%s Job execution failed.", jobName));
        }
    }
    
    private static JobParameters createJobParams() {
        return new JobParametersBuilder().addDate("date", new Date()).toJobParameters();
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMxOTQyNQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191125111310803.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMxOTQyNQ==,size_16,color_FFFFFF,t_70