我可以将 FlatfileItemReader 与 Taskexecutor 一起使用吗？答案

【问题标题】：Can I use FlatfileItemReader with Taskexecutor?我可以将 FlatfileItemReader 与 Taskexecutor 一起使用吗？
【发布时间】：2023-03-03 14:54:01
【问题描述】：

我可以在 Spring Batch 中将 FlatfileItemReader 与 Taskexecutor 一起使用吗？

我已经用 ThreadPoolTaskExecutor 实现了 FlatFileItemReader。当我在 ItemProcessor 中打印记录时，我没有得到一致的结果，即并非所有记录都被打印，有时其中一条记录被打印多次。它让我知道 FlatFileItemReader 不是线程安全的，而且它在 spring 文档中也这么说，但我看到一些博客说可以将 FlatFileItemReader 与 Task Executor 一起使用。

所以我的问题是：是否可以将 FlatfileItemReader 与 Task Executor 一起使用？

    @Bean
    @StepScope
    public FlatFileItemReader<DataLifeCycleEvent> csvFileReader(
            @Value("#{stepExecution}") StepExecution stepExecution) {

        Resource inputResource;
        FlatFileItemReader<DataLifeCycleEvent> itemReader = new FlatFileItemReader<>();

        itemReader.setLineMapper(new OnboardingLineMapper(stepExecution));
        itemReader.setLinesToSkip(1);
        itemReader.setSaveState(false);
        itemReader.setSkippedLinesCallback(new OnboardingHeaderMapper(stepExecution));
        String inputResourceString = stepExecution.getJobParameters().getString("inputResource");
        inputResource = new FileSystemResource(inputFileLocation + ApplicationConstant.SLASH + inputResourceString);
        itemReader.setResource(inputResource);
        stepExecution.getJobExecution().getExecutionContext().putInt(ApplicationConstant.ERROR_COUNT, 0);
        return itemReader;
    }

【问题讨论】：

你为什么需要那个？你的意思是使用面向块的步骤的任务执行器吗？请在不参考 Spring Batch 以了解您的要求的情况下解释您要实现的目标。
嗨@MahmoudBenHassine，我的要求是我想在多个线程中上传一个平面文件，比如每个线程需要100条记录（例如），验证记录，如果验证成功，写将其写入数据库，否则将其写入错误文件。
在春季批处理方面，我已经使用 FlatFileItemReader 实现了这一点，以读取文件并在我的步骤中添加任务执行器以启用多线程。当我在项目处理器中打印记录时，每次都会给出不一致的结果，一些记录被跳过，而另一些则重复打印。

标签： java spring multithreading spring-batch flatfilereader

【解决方案1】：

FlatFileItemReader 扩展了AbstractItemCountingItemStreamItemReader，即NOT thread-safe。所以如果你在多线程步骤中使用它，你需要同步它。

您可以将其包装在 SynchronizedItemStreamReader 中。这是一个简单的例子：

@Bean
public SynchronizedItemStreamReader<DataLifeCycleEvent> itemReader() {
    FlatFileItemReader<DataLifeCycleEvent> itemReader = ... // your item reader

    SynchronizedItemStreamReader<DataLifeCycleEvent> synchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
    synchronizedItemStreamReader.setDelegate(itemReader);
    return synchronizedItemStreamReader;
}

【讨论】：

感谢@Mahmoud Ben Hasine。这解决了这个问题。为了决定天气我应该将记录发送到错误文件写入器或数据库写入器，我实现了 ClassifyCompositeItemWriter 但声纳抛出错误，即 ClassifyCompositeItemWriter 内部使用的写入器也应该序列化，并且这些写入器中的所有其他对象。我第一次在 Spring 中使用 ClassifyCompositeItemWriter 遇到了这个问题，这导致我改变了我的完整实现并转移到了 ItemStreamWriter。你有解决这个问题的办法吗？

【解决方案2】：

这个方法给出了这个异常：-java.lang.ClassCastException: com.sun.proxy.$Proxy344 cannot be cast to org.springframework.batch.item.support.SynchronizedItemStreamReader

@Bean
public SynchronizedItemStreamReader<DataLifeCycleEvent> itemReader() {
    FlatFileItemReader<DataLifeCycleEvent> itemReader = ... // your item reader

    SynchronizedItemStreamReader<DataLifeCycleEvent> synchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
    synchronizedItemStreamReader.setDelegate(itemReader);
    return synchronizedItemStreamReader;
}

【讨论】：