Spring批处理：添加跳过限制时taskExecutor异常

Question

我正在使用Spring 4.0.5和Spring Batch 3.0.1

我有这样一个简单的步骤，它工作的很完美：

<step id="myStep" next="nextStep">
    <tasklet transaction-manager="myTxManager" task-executor="myTaskExecutor" throttle-limit="10">
        <batch:chunk reader="myItemReader" processor="myPDFItemProcessor" writer="myItemWriter" commit-interval="20">
        </batch:chunk>
    </tasklet>
</step>

我尝试了一个简单的跳过限制示例（配置跳过逻辑），如下所示：

<step id="myStep" next="nextStep">
    <tasklet transaction-manager="myTxManager" task-executor="myTaskExecutor" throttle-limit="10">
        <batch:chunk reader="myItemReader" processor="myPDFItemProcessor" writer="myItemWriter" commit-interval="20" skip-limit="10000000">
            <batch:skippable-exception-classes>
                <batch:include class="java.lang.Exception" />
            </batch:skippable-exception-classes>
        </batch:chunk>
    </tasklet>
</step>

2015-03-24 16:03:50 [WARN ] [org.springframework.batch.core.step.builder.FaultTolerantStepBuilder.detectStreamInReader(FaultTolerantStepBuilder.java:504)] Asynchronous TaskExecutor detected with ItemStream reader.  This is probably an error, and may lead to incorrect restart data being stored.
2015-03-24 16:04:18 [WARN ] [org.springframework.batch.core.step.item.ChunkMonitor.open(ChunkMonitor.java:118)] No ItemReader set (must be concurrent step), so ignoring offset data.
2015-03-24 16:04:18 [WARN ] [org.springframework.batch.core.step.item.ChunkMonitor.getData(ChunkMonitor.java:155)] ItemStream was opened in a different thread.  Restart data could be compromised.

任何帮助都非常感谢。谢谢

Answer 1

最后，我可以通过Samwise的提示来解决我的问题。

当您在多线程步骤中有一个ItemReader时，为了使该读取线程安全，代码如下：

public SynchronizedItemReader<T> implements ItemReader<T> {
  private final ItemReader<T> delegate; 
  public SynchronizedItemReader(ItemReader<T> delegate) {
    this.delegate = delegate;
  }
  public synchronized T read () {
    return delegate.read();
  }
}

该类中的委托ItemReader是您希望使其成为线程安全的当前ItemReader。