Spring Batch：聚合阅读器/作者问题

Question

我正在尝试使用 Spring 批处理并实现聚合读取器（批处理文件，其中多个记录在写入时应被视为一条记录）。这是我的读者的代码 sn-p：

public class AggregatePeekableReader implements ItemReader<List<T>>, ItemStream {



    private SingleItemPeekableItemReader<T> reader;



    private boolean process(T currentRecord , InvoiceLineItemsHolder holder) throws UnexpectedInputException, ParseException, Exception {

        next = peekNextInvoiceRecord();

        // finish processing if we hit the end of file
        if (currentRecord == null ) {
                LOG.info("Exhausted ItemReader ( END OF FILE)");
                holder.exhausted = true;
                return false;
        }

        if ( currentRecord.hasSameInvoiceNumberAndVendorNumber(next)){
                LOG.info("Found new line item to current invocie record");
                holder.records.add(currentRecord);
                currentRecord = null;
                return true;
        }else{ 

            holder.records.add(currentRecord);
                return false;           
        }

}

    private T getNextInvoiceRecord () {

        T record=null;

        try {
            record=reader.read();
        } catch (UnexpectedInputException e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);
            throw e;
        } catch (ParseException e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);
            throw e;
        } catch (Exception e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);


        }

        return record;
    }

    private T peekNextInvoiceRecord() {

        T next=null;

        try {
            next=reader.peek();
        } catch (UnexpectedInputException e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);
            throw e;
        } catch (ParseException e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);
            throw e;
        } catch (Exception e) {
            ALERT.error(LogMessageFormatter.format(Severity.HIGH,
                    BATCH_FILE_READ_EXCEPTION, e), e);
        }
        return next;
    }

    public   void close () {
        reader.close();
    }

    public SingleItemPeekableItemReader<T> getReader() {
        return reader;
    }


    public   void setReader(SingleItemPeekableItemReader<T> reader) {
        this.reader = reader;
    }

    private class InvoiceLineItemsHolder {
        List<T> records = new ArrayList<T>();

        boolean exhausted = false;
}


    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        // 
        reader.open(executionContext);

    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        // TODO 

    }

    @Override
    public List<T> read() throws Exception, UnexpectedInputException, ParseException,
            NonTransientResourceException {
        CLASS holder = new SOMECLASS()

        synchronized (this) {

           while (process(getNextInvoiceRecord(), holder)) {
                continue;
            }
            if (!holder.exhausted) {

                return holder.records;
            } else {
                //When you hit the end of the file,close the reader.
                close();
                return null;
            }

        }

    }

}

上面是一个实现可窥视阅读器的工作示例。这会窥视下一行 （不读取）并确定是否到达逻辑行尾（有时 多行可以组成一个事务）

Answer 1

您需要为阅读器实现ItemStream 接口。这将提示 Spring Batch，您的阅读器需要一些操作来打开/关闭流：

public class InvoiceLineItemAggregatePeekableReader extends AbstractItemStreamItemReader<List<SAPInvoicePaymentRecord>> {

    @Override
    public void close() {
    ...
    }
}

无论步骤执行期间发生什么错误，流都会关闭。有关更多示例，请检查 Spring Batch 本身的类（例如 FlatFileItemReader）。

Answer 2

我无法将输入文件移动到错误文件夹，因为阅读器 没有关闭

您可以复制文件并在旧文件上使用File.deleteOnExit() 以便以后删除，或者在额外的步骤中删除旧文件，例如使用一个简单的 tasklet 和一个仅在业务步骤出现异常时才调用 deleteTaskletStep 的流程