春季批：汇总读者/作家问题 [英] Spring Batch :Aggregated reader / writer Issue

问题描述

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

  public class AggregatePeekableReader implements ItemReader< List< T>> ItemStream {
 
 
 
 private SingleItemPeekableItemReader< T>读者; 
 
 
 
私有布尔进程（T currentRecord，InvoiceLineItemsHolder holder）throws UnexpectedInputException，ParseException，Exception {
 
 next = peekNextInvoiceRecord（）; 
 
 //完成处理，如果我们命中文件
的结尾if（currentRecord == null）{
 LOG.info（Exhausted ItemReader（END OF FILE））; 
 holder.exhausted = true; 
返回false; 
} 
 
 if（currentRecord.hasSameInvoiceNumberAndVendorNumber（next））{
 LOG.info（找到新行项目到当前invocie记录）; 
 holder.records.add（currentRecord）; 
 currentRecord = null; 
返回true; 
} else {
 
 holder.records.add（currentRecord）; 
返回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）; 
 
 
} 
 
返回记录; 
} 
 
 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; 
} 
 
私有类InvoiceLineItemsHolder {
列表< 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 {
 //当您点击文件的末尾时，关闭阅读器。 
 close（）; 
返回null; 
} 
 
} 
 
} 
  

}

上面是一个实现可窥视读取器的工作示例。这个视图看下一行
（不读取它）并确定逻辑结束线路达到（某些时候
多行可以组成单个事务）

解决方案

您需要实现 ItemStream 界面为读者。这将提示Spring Batch，您的读者需要一些操作才能打开/关闭流：

  public class InvoiceLineItemAggregatePeekableReader extends AbstractItemStreamItemReader<列表与LT; SAPInvoicePaymentRecord>> {
 
 @Override 
 public void close（）{
 ... 
} 
} 
  / pre> 
 
 
在执行步骤期间发生的任何错误都会关闭流。有关更多示例，请查看Spring Batch本身的类（例如 FlatFileItemReader ）。
 
I am trying to use Spring batch and implement an aggregated reader (batch file, where multiple records should be treated as one record while writing). Here is the code snippet for my reader:
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;
            }

        }

    }
}

The above is a working example for implementing a peekable reader.This peeks the next line
(doesnt read it) and determines whether  a logical end of line is reached (some times
multiple lines can make up a single transaction)
解决方案
You need to implement ItemStream interface for reader. This will give a hint to Spring Batch, that your reader requires some actions to open/close a stream:
public class InvoiceLineItemAggregatePeekableReader extends AbstractItemStreamItemReader<List<SAPInvoicePaymentRecord>> {

    @Override
    public void close() {
    ...
    }
}
Streams are closed whatever error occurred during step execution. For more examples check classes from Spring Batch itself (e.g. FlatFileItemReader).

                        
这篇关于春季批：汇总读者/作家问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！