如何实现并行处理以提高处理时间

Question

目前我正在使用 springbatch 以下列方式处理 csv 和 excel 文件。

1. Reader（将解析 csv/excel 文件和男性 pojo）
2. 处理器（无论该记录是否存在于 DB 中都会命中 Db）
3. Writer（将pojo推送到消息队列）

实时我有 50k + 记录要处理，我的代码几乎需要 25 分钟。我想通过实现并行处理来缩短处理时间（这样我们就可以在更短的时间内并行处理）。

但我不知道如何使用 Spring Batch 实现并行处理。任何人都可以指导我如何做到这一点或任何建议以改善处理时间。

@Bean
    public TaskExecutor taskExecutor(){
        return new SimpleAsyncTaskExecutor("CSV-Async-batch");
    }


    @Bean(name="csvjob")
    public Job job(JobBuilderFactory jobBuilderFactory,StepBuilderFactory stepBuilderFactory,ItemReader<List<CSVPojo>> itemReader,ItemProcessor<List<CSVPojo>,CsvWrapperPojo> itemProcessor,AmqpItemWriter<CsvWrapperPojo> itemWriter){
        Step step=stepBuilderFactory.get("ETL-CSV").<List<CSVPojo>,CsvWrapperPojo>chunk(100)
                .reader(itemReader)
                .processor(itemProcessor)
                .writer(itemWriter)
                .taskExecutor(taskExecutor())
                .throttleLimit(40)
                .build();



        Job csvJob= jobBuilderFactory.get("ETL").incrementer(new RunIdIncrementer())
        .start(step).build();

====SynchronizedItemStreamReader 的阅读器==================

@Component
public class Reader extends SynchronizedItemStreamReader<List<CSVPojo>> {

    public static MultipartFile reqFile=null;
    List<CSVPojo> result = new ArrayList<CSVPojo>();

    @Autowired
    private CSVProcessService csvProcessService;

    public static boolean batchJobState ;

    /*public Reader(MultipartFile file){

        this.reqFile=file;
    }*/

    public void setDelegate(ItemStreamReader<List<CSVPojo>> delegate){

        /*try {
            this.read();
        } catch (UnexpectedInputException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (ParseException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (NonTransientResourceException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }*/
    }


    @Override
    public List<CSVPojo> read() throws Exception, UnexpectedInputException,
            ParseException, NonTransientResourceException {
        // TODO Auto-generated method stub
        if(!batchJobState){
        result=csvProcessService.processCSVFile(reqFile);
        System.out.println("in batch job reader");
        batchJobState=true;
        return result;
        }
        return null;
    }

}

提前致谢！！！

Answer 1

您可以使用分区技术对输入文件进行分区并并行处理它们。这在参考文档的Partitioning 部分中有详细说明。

您还可以查看spring-batch-samples 模块中的Local partitioning sample 和Remote partitioning sample。

这个问题也有类似的问题，我这里补充一下供参考：

• Spring batch- Parallel processing
• Best Spring batch scaling strategy

希望这会有所帮助。