使用SpringBatch读取csv文件

我就是我 2024-04-07 11:13 74阅读 0赞

#### 文章目录 ####

*  1、需求
 *  2、解决方案
 *  3、注意事项
 *   *  1、文件路径的获取
     *  2、各个Step如果获取到ExecutionContext中的值
     *  3、FlatFileItemReader使用注意
 *  4、实现步骤
 *   *  1、导入依赖，配置
     *   *  1、导入依赖
         *  2、初始化SpringBatch数据库
     *  2、构建文件读取路径
     *  3、构建Tasklet，输出文件路径
     *  4、编写实体类
     *  5、编写Job配置
     *  6、编写Job启动类
     *  7、自动配置SpringBatch
 *  5、执行结果
 *  6、完整代码

## 1、需求 ##

系统每日从某个固定的目录中读取`csv`文件，并在控制台上打印。

## 2、解决方案 ##

要解决上述需求，可以使用的方法有很多，此处选择使用`Spring Batch`来实现。

## 3、注意事项 ##

### 1、文件路径的获取 ###

此处简单处理，读取 `JobParameters` 中的日期，然后构建一个文件路径，并将文件路径放入到 `ExecutionContext`中。此处为了简单，文件路径会在程序中写死，但是同时也会将文件路径存入到 `ExecutionContext` 中，并且在具体的某个`Step`中从`ExecutionContext`中获取路径。

**注意：**  
`ExecutionContext`中存入的数据虽然在各个`Step`中都可以获取到，但是`不推荐`存入`比较大`的数据到`ExecutionContext`中，因为这个对象的数据需要存入到数据库中。

### 2、各个Step如果获取到ExecutionContext中的值 ###

1.  类上加入 `@StepScope` 注解
2.  通过 `@Value("#{jobExecutionContext['importPath']}")` 来获取

**eg:**

@Bean
    @StepScope
    public FlatFileItemReader<Person> readCsvItemReader(@Value("#{jobExecutionContext['importPath']}") String importPath) {
        
        // 读取数据
        return new FlatFileItemReaderBuilder<Person>()
                .name("read-csv-file")
                .resource(new ClassPathResource(importPath))
                .delimited().delimiter(",")
                .names("username", "age", "sex")
                .fieldSetMapper(new RecordFieldSetMapper<>(Person.class))
                .build();
    }

解释：在程序实例化`FlatFileItemReader`的时候，此时是没有jobExecutionContext的，那么就会报错，如果加上`@StepScope`，此时就没有问题了。`@StepScope`表示到达Step阶段才实例化这个Bean

### 3、FlatFileItemReader使用注意 ###

当我们使用`FlatFileItemReader`来读取我们的csv文件时，此处需要返回 `FlatFileItemReader`类型，而不能直接返回`ItemReader`，否则可能出现如下错误 `Reader must be open before it can be read`

## 4、实现步骤 ##

### 1、导入依赖，配置 ###

#### 1、导入依赖 ####

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

#### 2、初始化SpringBatch数据库 ####

spring.datasource.username=root
    spring.datasource.password=root@1993
    spring.datasource.url=jdbc:mysql://127.0.0.1:3306/spring-batch?useUnicode=true&characterEncoding=utf8&autoReconnectForPools=true&useSSL=false
    spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
    
    # 程序启动时，默认不执行job
    spring.batch.job.enabled=false
    spring.batch.jdbc.initialize-schema=always
    # 初始化spring-batch数据库脚本
    spring.batch.jdbc.schema=classpath:org/springframework/batch/core/schema-mysql.sql

### 2、构建文件读取路径 ###

此处我的想法是，在`JobExecutionListener`中完成文件路径的获取，并将之放入到`ExecutionContext`，然后在各个`Step`中就可以获取到文件路径的值了。

/**
     * 在此监听器中，获取到具体的需要读取的文件路径，并保存到 ExecutionContext
     *
     * @author huan.fu
     * @date 2022/8/30 - 22:22
     */
    @Slf4j
    public class AssemblyReadCsvPathListener implements JobExecutionListener {
        
        @Override
        public void beforeJob(JobExecution jobExecution) {
        
            ExecutionContext executionContext = jobExecution.getExecutionContext();
            JobParameters jobParameters = jobExecution.getJobParameters();
            String importDate = jobParameters.getString("importDate");
            log.info("从 job parameter 中获取的 importDate 参数的值为:[{}]", importDate);
            String readCsvPath = "data/person.csv";
            log.info("根据日期组装需要读取的csv路径为:[{}],此处排除日期，直接写一个死的路径", readCsvPath);
            executionContext.putString("importPath", readCsvPath);
        }
    
        @Override
        public void afterJob(JobExecution jobExecution) {
        
    
        }
    }

### 3、构建Tasklet，输出文件路径 ###

@Slf4j
    @Component
    @StepScope
    public class PrintImportFilePathTaskLet implements Tasklet {
        
    
        @Value("#{jobExecutionContext['importPath']}")
        private String importFilePath;
    
        @Value("#{jobParameters['importDate']}")
        private String importDate;
    
        @Override
        public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        
    
            log.info("从job parameter 中获取到的 importDate:[{}],从 jobExecutionContext 中获取的 importPath:[{}]",
                    importDate, importFilePath);
    
            return RepeatStatus.FINISHED;
        }
    }

需要注意的是，此类上加入了 `@StepScope`注解

### 4、编写实体类 ###

@AllArgsConstructor
    @Getter
    @ToString
    public class Person {
        
        /**
         * 用户名
         */
        private String username;
        /**
         * 年龄
         */
        private Integer age;
        /**
         * 性别
         */
        private String sex;
    }

### 5、编写Job配置 ###

@Configuration
    @AllArgsConstructor
    @Slf4j
    public class ImportPersonJobConfig {
        
    
        private final JobBuilderFactory jobBuilderFactory;
        private final StepBuilderFactory stepBuilderFactory;
    
        private final PrintImportFilePathTaskLet printImportFilePathTaskLet;
        private final ItemReader<Person> readCsvItemReader;
    
        @Bean
        public Job importPersonJob() {
        
            // 获取一个job builder, jobName可以是不存在的
            return jobBuilderFactory.get("import-person-job")
                    // 添加job execution 监听器
                    .listener(new AssemblyReadCsvPathListener())
                    // 打印 job parameters 和 ExecutionContext 中的值
                    .start(printParametersAndContextVariables())
                    // 读取csv的数据并处理
                    .next(handleCsvFileStep())
                    .build();
        }
    
        /**
         * 读取数据
         * 注意：此处需要返回 FlatFileItemReader类型，而不要返回ItemReader
         * 否则可能报如下异常 Reader must be open before it can be read
         *
         * @param importPath 文件路径
         * @return reader
         */
        @Bean
        @StepScope
        public FlatFileItemReader<Person> readCsvItemReader(@Value("#{jobExecutionContext['importPath']}") String importPath) {
        
            // 读取数据
            return new FlatFileItemReaderBuilder<Person>()
                    .name("read-csv-file")
                    .resource(new ClassPathResource(importPath))
                    .delimited().delimiter(",")
                    .names("username", "age", "sex")
                    .fieldSetMapper(new RecordFieldSetMapper<>(Person.class))
                    .build();
        }
    
        @Bean
        public Step handleCsvFileStep() {
        
    
            // 每读取一条数据，交给这个处理
            ItemProcessor<Person, Person> processor = item -> {
        
                if (item.getAge() > 25) {
        
                    log.info("用户[{}]的年龄:[{}>25]不处理", item.getUsername(), item.getAge());
                    return null;
                }
                return item;
            };
    
            // 读取到了 chunk 大小的数据后，开始执行写入
            ItemWriter<Person> itemWriter = items -> {
        
                log.info("开始写入数据");
                for (Person item : items) {
        
                    log.info("{}", item);
                }
            };
    
            return stepBuilderFactory.get("handle-csv-file")
                    // 每读取2条数据，执行一次write，当每read一条数据后，都会执行process
                    .<Person, Person>chunk(2)
                    // 读取数据
                    .reader(readCsvItemReader)
                    // 读取一条数据就开始处理
                    .processor(processor)
                    // 当读取的数据的数量到达 chunk 时，调用该方法进行处理
                    .writer(itemWriter)
                    .build();
        }
    
        /**
         * 打印 job parameters 和 ExecutionContext 中的值
         * <p>
         * TaskletStep是一个非常简单的接口，仅有一个方法——execute。
         * TaskletStep会反复的调用这个方法直到获取一个RepeatStatus.FINISHED返回或者抛出一个异常。
         * 所有的Tasklet调用都会包装在一个事物中。
         *
         * @return Step
         */
        private Step printParametersAndContextVariables() {
        
            return stepBuilderFactory.get("print-context-params")
                    .tasklet(printImportFilePathTaskLet)
                    // 当job重启时，如果达到了3此，则该step不在执行
                    .startLimit(3)
                    // 当job重启时，如果该step的是已经处理完成即COMPLETED状态时，下方给false表示该step不在重启，即不在执行
                    .allowStartIfComplete(false)
                    // 添加 step 监听
                    .listener(new CustomStepExecutionListener())
                    .build();
        }
    }

### 6、编写Job启动类 ###

@Component
    @Slf4j
    public class StartImportPersonJob {
        
    
        @Autowired
        private Job importPersonJob;
        @Autowired
        private JobLauncher jobLauncher;
    
        @PostConstruct
        public void startJob() throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException {
        
            JobParameters jobParameters = new JobParametersBuilder()
                    .addString("importDate", LocalDate.of(2022, 08, 31).format(DateTimeFormatter.ofPattern("yyyyMMdd")))
                    .toJobParameters();
            JobExecution execution = jobLauncher.run(importPersonJob, jobParameters);
            log.info("job invoked");
        }
    }

### 7、自动配置SpringBatch ###

@SpringBootApplication
    @EnableBatchProcessing
    public class SpringBatchReadCsvApplication {
        
    
        public static void main(String[] args) {
        
            SpringApplication.run(SpringBatchReadCsvApplication.class, args);
        }
    }

主要是 `@EnableBatchProcessing` 注解

## 5、执行结果 ##

![执行结果][3a4fa6db2ec341fa947e55ef7317e9f5.png]

## 6、完整代码 ##

[https://gitee.com/huan1993/spring-cloud-parent/tree/master/spring-batch/spring-batch-read-csv][https_gitee.com_huan1993_spring-cloud-parent_tree_master_spring-batch_spring-batch-read-csv]

[3a4fa6db2ec341fa947e55ef7317e9f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/8b5aceb2307349adae3c0c4a1c40dbcd.png
[https_gitee.com_huan1993_spring-cloud-parent_tree_master_spring-batch_spring-batch-read-csv]: https://gitee.com/huan1993/spring-cloud-parent/tree/master/spring-batch/spring-batch-read-csv