Spring Boot批处理-MultiResourceItemReader：出错时移至下一个文件

Question

在批处理服务中，我使用 MultiResourceItemReader 读取多个 XML 文件，它委托给 StaxEventItemReader。

如果在读取文件时引发错误（例如解析异常），我想指定 Spring 开始读取下一个匹配文件。例如，使用 @OnReadError 注释和/或 SkipPolicy。

目前，当引发读取异常时，批处理会停止。

有人知道怎么做吗？

编辑：我看到 MultiResourceItemReader 有一个方法 readNextItem()，但它是私有的 -_-

Answer 1

我暂时没有使用 SB，但查看 MultiResourceItemReader 代码我想您可以编写自己的 ResourceAwareItemReaderItemStream 包装器，在其中检查设置为移动到下一个文件或执行标准读取的标志委托。
此标志可以存储到执行上下文或包装器中，并且应该在下一步移动后清除。

class MoveNextReader<T> implements ResourceAwareItemReaderItemStream<T> {
  private ResourceAwareItemReaderItemStream delegate;
  private boolean skipThisFile = false;

  public void setSkipThisFile(boolean value) {
    skipThisFile = value;
  }

  public void setResource(Resource resource) {
    skipThisFile = false;
    delegate.setResource(resource);
  }

  public T read() {
    if(skipThisFile) {
      skipThisFile = false;
      // This force MultiResourceItemReader to move to next resource
      return null;
    }
    return delegate.read();
  }
}

将此类用作MultiResourceItemReader 和@OnReadErrorinject MoveNextReader 的委托并设置MoveNextReader.skipThisFile。

我无法自己测试代码，但我希望这是一个很好的起点。

Answer 2

这是我的最终类，用于读取多个 XML 文件并在一个文件发生读取错误时跳转到下一个文件（感谢 Luca 的想法）。

我的自定义 ItemReader，从 MultiResourceItemReader 扩展：

public class MyItemReader extends MultiResourceItemReader<InputElement> {

    private SkippableResourceItemReader<InputElement> reader;

    public MyItemReader() throws IOException {
        super();

        // Resources
        PathMatchingResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver();
        this.setResources( resourceResolver.getResources( "classpath:input/inputFile*.xml" ) );

        // Delegate reader
        reader = new SkippableResourceItemReader<InputElement>();
        StaxEventItemReader<InputElement> delegateReader = new StaxEventItemReader<InputElement>();
        delegateReader.setFragmentRootElementName("inputElement");
        Jaxb2Marshaller unmarshaller = new Jaxb2Marshaller();
        unmarshaller.setClassesToBeBound( InputElement.class );
        delegateReader.setUnmarshaller( unmarshaller );
        reader.setDelegate( delegateReader );

        this.setDelegate( reader );
    }

    [...] 

    @OnReadError
    public void onReadError( Exception exception ){
        reader.setSkipResource( true );
    }
}

中间的 ItemReader 用于跳过当前资源：

public class SkippableResourceItemReader<T> implements ResourceAwareItemReaderItemStream<T> {

    private ResourceAwareItemReaderItemStream<T> delegate;
    private boolean skipResource = false;

    @Override
    public void close() throws ItemStreamException {
        delegate.close();
    }

    @Override
    public T read() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {
        if( skipResource ){
            skipResource = false;
            return null;
        }
        return delegate.read();
    }

    @Override
    public void setResource( Resource resource ) {
        skipResource = false;
        delegate.setResource( resource );
    }

    @Override
    public void open( ExecutionContext executionContext ) throws ItemStreamException {
        delegate.open( executionContext );
    }

    @Override
    public void update( ExecutionContext executionContext ) throws ItemStreamException {
        delegate.update( executionContext );
    }

    public void setDelegate(ResourceAwareItemReaderItemStream<T> delegate) {
        this.delegate = delegate;
    }

    public void setSkipResource( boolean skipResource ) {
        this.skipResource = skipResource;
    }
}