water-iot/meter-reading-common/src/main/java/com/huaxu/zoniot/config/JobConfiguration.java

package com.huaxu.zoniot.config;

import com.huaxu.zoniot.common.CommonFileItemReader;
import com.huaxu.zoniot.common.CommonMybatisItemWriter;
import com.huaxu.zoniot.common.Constants;
import com.huaxu.zoniot.common.ServiceException;
import com.huaxu.zoniot.dto.WaterMigrateFileItemDto;
import com.huaxu.zoniot.entity.WaterMeter;
import com.huaxu.zoniot.entity.WaterMigrateFileItem;
import com.huaxu.zoniot.listener.MigrateJobListener;
import com.huaxu.zoniot.listener.MigrateJobSkipListener;
import com.huaxu.zoniot.processor.MigrateFileItemProcessor;
import com.huaxu.zoniot.writer.WaterMeterWriter;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.batch.core.*;
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.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;

import java.util.Map;

/**
 * <p></p>
 *
 * @Author wilian.peng
 * @Date 2021/1/12 18:24
 * @Version 1.0
 */
@Configuration
@EnableBatchProcessing
public class JobConfiguration implements StepExecutionListener {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
    private MigrateFileItemProcessor  migrateFileItemProcessor;

    @Autowired
    private MigrateJobSkipListener migrateJobSkipListener ;

    @Autowired
    private WaterMeterWriter waterMeterWriter;

    @Autowired
    private MigrateJobListener migrateJobListener;

    @Value("${spring.batch.chunk.size}")
    private Integer chunkSize;

    Map<String, JobParameter> parameterMap;

    @Bean
    public Job migrateJob() {
        return jobBuilderFactory.get("migrateJob")
                .start(waterMeterMigrateStep())
                .listener(migrateJobListener)
                .build();
    }

    @Bean
    public Step waterMeterMigrateStep() {
        return stepBuilderFactory.get("waterMeterMigrateStep")
                .listener(this)
                .<WaterMigrateFileItem, WaterMeter>chunk(chunkSize)
                .reader(migrateFileItemItemReader())
                .processor(migrateFileItemProcessor)
                .writer(waterMeterWriter)
                .faultTolerant()
                .skip(ServiceException.class)
                .skipLimit(Integer.MAX_VALUE)
                .listener(migrateJobSkipListener)
                .taskExecutor(taskExecutor())
                .throttleLimit(8)
                .build();
    }

    @Bean
    public TaskExecutor taskExecutor(){
        return new SimpleAsyncTaskExecutor("spring_batch");
    }

    @Bean
    @StepScope
    public CommonFileItemReader<WaterMigrateFileItemDto> migrateFileItemItemReader() {
        String filePath = parameterMap.get(Constants.FILE_FULL_PATH_TAG).getValue().toString();
        return new CommonFileItemReader<>(WaterMigrateFileItemDto.class,filePath);
    }


    @Override
    public void beforeStep(StepExecution stepExecution) {
        parameterMap = stepExecution.getJobParameters().getParameters();
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }
}