Spring Batch“使用ItemStream阅读器检测到异步TaskExecutor”

Question

首先，这个问题似乎几乎正是我们想要做的： Parallel step execution of ItemStreamReader in SpringBatch

Spring batch 3.0.10，在 Java EE 服务器（WebSphere、Java 8、Java EE 6）下运行。我也不是这段代码的原作者。

XML 配置，我们有一个批处理步骤，从文件中读取一行，处理该行（我们想要多线程的部分）并将一条记录写入数据库：

    <batch:step id="processRenewalsStep" next="saveResponseFileStep">
        <batch:tasklet task-executor="taskExecutor" throttle-limit="4">
            <batch:chunk
                reader="batchRenewalCsvFileItemReader"
                writer="asyncBatchRenewalDb2ItemWriter"
                processor="asyncBatchRenewalProcessor"
                commit-interval="1"
                skip-limit="10">
                <batch:skippable-exception-classes>
                    <batch:include class="java.lang.Exception"/>
                </batch:skippable-exception-classes>
                <batch:listeners>
                    <batch:listener ref="batchSkipListener"/>
                </batch:listeners>
            </batch:chunk>
        </batch:tasklet>
    </batch:step>

即使关注the suggested aynchronous wrapping in the accepted answer there:

<bean id="asyncBatchRenewalProcessor" class="org.springframework.batch.integration.async.AsyncItemProcessor"
      p:taskExecutor-ref="taskExecutor"
      p:delegate-ref="batchRenewalProcessor"/>

<bean id="asyncBatchRenewalDb2ItemWriter" class="org.springframework.batch.integration.async.AsyncItemWriter"
      p:delegate-ref="batchRenewalDb2ItemWriter"/>

我们的启动日志仍然包含：

22:25:25,252（默认值：3）警告 org.springframework.batch.core.step.builder.FaultTolerantStepBuilder：使用 ItemStream 读取器检测到异步 TaskExecutor。这可能是一个错误，并可能导致存储不正确的重启数据。

和

22:34:03,755 (WorkManager.DefaultWorkManager : 0) WARN org.springframework.batch.core.step.item.ChunkMonitor:No ItemReader set (must be concurrent step)，所以忽略偏移数据。

还有几个例子：

22:34:03,880 (WorkManager.DefaultWorkManager : 2) 警告 org.springframework.batch.core.step.item.ChunkMonitor:ItemStream 在不同的线程中打开。重启数据可能会受到影响。

我认为这是有道理的，因为我们包装了处理器和写入器，但没有包装阅读器的类？

运行我的工作，日志记录确实表明正在使用来自我的容器 Executor 的不同线程，但我在使用异步包装器之前没有尝试过，所以即使没有它，它可能也会“工作”？

那么有什么方法可以防止警告，以及他们警告的情况吗？

更新：基于this in the documentation：

如果读取器不是线程安全的，在您自己的同步委派器中使用它可能仍然有效。您可以同步对 read() 的调用，只要处理和写入是块中最昂贵的部分，您的步骤仍然可以比单线程配置更快地完成。

我尝试创建一个 FlatFileItemReader 子类，它确实同步它自己的 read() 方法，该方法调用底层的 read() 方法，但我仍然收到 WARN 日志记录。我不知道我是否误解了该建议，或者我现在所做的是否安全但无法被检测为安全，所以它仍在记录。

Answer 1

FlatFileItemReader 不是线程安全的，因为它扩展了AbstractItemCountingItemStreamItemReader，即not thread-safe。所以在多线程步骤中使用它是不正确的（否则执行上下文数据可能会被并发线程破坏，因此会出现警告）。

为避免任何并发问题，您可以尝试对文件进行分区（物理上或逻辑上）并使每个线程在不同的分区上工作。