使用Spring批处理将多个csv文件合并为单个csv

Question

我有一个商业案例，使用Spring batch将多个csv文件（每个文件大约1000个，包含1000条记录）合并成单个csv。

请帮助我提供方法和性能方面的指导和解决方案。

到目前为止，我已经尝试了两种方法，

方法1。

Tasklet chunk与multiResourceItemReader一起从目录中读取文件，FlatFileItemWriter作为项目编写器。

这里的问题是，它的处理速度非常慢，因为这是单线程的，但这种方法按预期工作。

方法2：使用MultiResourcePartitioner partitioner和AsynTaskExceutor作为任务执行器。

这里的问题是，由于它是异步多线程，数据在合并到最终的单个文件时被覆盖/损坏。

Answer 1

由于源文件和目标文件之间的标题很常见，我不建议使用Spring Batch提供的读取器将行转换为特定的bean，因为不需要列级别的信息

而且，每个文件的分区速度会非常慢（如果有那么多文件）

我的建议是编写自定义阅读器

如果您有足够的内存来保存所有文件中的行，那么您可以一次读取所有文件

Answer 2

您可以将FlatFileItemWriter包装在AsyncItemWriter中，并与AsyncItemProcessor一起使用。这不会损坏数据并提高性能，因为处理和写入将通过多个线程进行。

@Bean
    public AsyncItemWriter asyncItemWriter() throws Exception {
        AsyncItemWriter<Customer> asyncItemWriter = new AsyncItemWriter<>();

        asyncItemWriter.setDelegate(flatFileItemWriter);
        asyncItemWriter.afterPropertiesSet();

        return asyncItemWriter;
    }

@Bean
    public AsyncItemProcessor asyncItemProcessor() throws Exception {
        AsyncItemProcessor<Customer, Customer> asyncItemProcessor = new AsyncItemProcessor();

        asyncItemProcessor.setDelegate(itemProcessor());
        asyncItemProcessor.setTaskExecutor(threadPoolTaskExecutor());
        asyncItemProcessor.afterPropertiesSet();

        return asyncItemProcessor;
    }

@Bean
    public TaskExecutor threadPoolTaskExecutor() {

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(10);
        executor.setThreadNamePrefix("default_task_executor_thread");
        executor.initialize();
        return executor;

    }