スプリングバッチ - プロセッサ内で複数のレコードを同時に処理する方法は？

Question

私は、レコードを解析して処理するファイルを持っています。それは行単位で（一度に1つのレコードを解析して）うまく動作しています。私の要件は、複数の行を解析し、各レコードから必要な情報を取得し、すべてのレコードから取得した情報を結合した後、ビジネスロジックを実行するサービスを呼び出します。私はプロセッサクラス内でこのロジックを実行する必要があります。データは、例えば以下のようにのように見える：

001 123456 987654321551580ウェインDR 1

001 123456 987654321552APT 786 1

001 123456 987654321553LOS ANGELES 1

001 123456 987654321554CA 1

データe列30-32で利用可能なものは、私が各レコードから取り出したいものです。上の例では、それぞれの行にそれぞれ551,552,553,554,555の値があります。彼らはすべて一緒にファイルに入っています。したがって、プロセッサの現在のアイテムが最初の行を解析し、その '551'（ビジネスコードのAddress Line1を意味する）を見つけたら、この行に続く残りのアドレスをフェッチして1つの完全な住所。最後に、このアドレスをプロセッサーからサービスクラスに渡してから、ファイル内の次のレコードを解析するために移動します。私の問題は、プロセッサがレコードごとに行ごとに動作することです。このようにして、これらすべての関連する行を追跡/関連付けることはできません。 私は簡単な方法で私の問題を説明することができない場合は申し訳ありません。私は春のバッチに新しいとまだ学んでいます。

Answer 1

ランダムに広がるのではなく、関連するデータレコードがファイル内で隣り合っていることがわかっている場合は、SingleItemPeekableItemReaderを利用して複数の行を関連付けて1つの完全なオブジェクトを作成できます。このolder answerにはもう少し詳しい情報があります。

例コンテキストファイル：

<bean id="flatFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader"> 
    <property name="resource" value="file:c:/temp/file.txt" /> 
    <property name="lineMapper"> 
     <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper"> 
      <property name="lineTokenizer" ref="yourTokenizer"/> 
      <property name="fieldSetMapper" ref="yourMapper"/> 
     </bean> 
    </property> 
</bean> 

例チラッと覗くリーダー：

public class YourPeekingReader extends SingleItemPeekableItemReader<YourObject> { 

    @Override 
    public YourObject read() { 

     YourObject item = super.read(); 

     if (item == null) { 
      return null; 
     } 

     while (true) { 
      YourObject possibleRelatedObject = peek(); 
      if (possibleRelatedObject == null) { 
       return item; 
      } 

      //logic to determine if next line in file relates to same object 
      boolean matches = false; 

      if (matches) { 
       item.addRelatedInfo(super.read()); 
      } else { 
       return item; 
      } 
     } 


    } 

} 

Answer 2

Dean..thanks @再び。私のコードでより正確には、ここでは

顧客レコード-reader.xmlあなたが見ることができるように、私はFlatFileItemReaderをラップするItemReaderのカスタム実装を使用していない

<batch:job id="myFileReaderJob"> 
    <batch:step id="stepA" next="stepSuccess"> 
     <batch:tasklet>   
     <batch:chunk reader="myInputReader" processor="myProcessor" writer="myWriter" commit-interval="1"/> 
     </batch:tasklet> 
    </batch:step> 
    <batch:step id="stepSuccess"> 
     <batch:tasklet ref="successTasklet" /> 
    </batch:step> 
</batch:job> 

<bean id="myInputReader" scope="step" class="org.springframework.batch.item.file.FlatFileItemReader"> 
     <property name="lineMapper" ref="myLineMapper" /> 
</bean> 

<bean id="myLineMapper" class="org.springframework.batch.item.file.mapping.DefaultLineMapper"> 
<property name="lineTokenizer"> 
<bean id="fixedLengthLineTokenizer" class="org.springframework.batch.item.file.transform.FixedLengthTokenizer"> 
     <property name="names" value="custRecord,tranId,partyId,uniquePartyId,deNum,deVal" /> 
     <property name="columns" value="1-75,1-3,6-11,21-29,30-32,33-62" /> 
     <property name="strict" value="false" /> 
    </bean> 
</property> 
<property name="fieldSetMapper"> 
    <bean  class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper"> 
     <property name="prototypeBeanName" value="myInputData" /> 
    </bean> 
</property> 
</bean> 

です。上記のコードでSingleItemPeekableItemReaderを実装する方法を変更する方法について詳しく説明してください。

おかげ

Answer 3

@Dean 
I tried implementing as per your suggestion 

Config1.xml 
<import resource="classpath*:/META-INF/java-batchlauncher/mainConfig.xml" /> 

<batch:job id="prT813FileReaderJob"> 
     <batch:step id="stepA" next="stepB"> 
      <batch:tasklet ref="aTasklet" /> 
     </batch:step> 
     <batch:step id="stepB" next="stepSuccess"> 
      <batch:tasklet>   
       <batch:chunk reader="prT813MultiReader" processor="participantRecordT813Processor" writer="prT813ItemWriter" commit-interval="1"/> 
       <batch:listeners> 
        <batch:listener ref="enabledFeaturesStepListener"/> 
       </batch:listeners> 
       <batch:transaction-attributes propagation="NEVER"/> 
      </batch:tasklet> 
     </batch:step> 
     <batch:step id="stepSuccess"> 
      <batch:tasklet ref="successTasklet" /> 
     </batch:step> 
    </batch:job> 

My mainConfig.xml file changes: 

<bean id="prT813MultiReader" scope="step" class="org.springframework.batch.item.file.MultiResourceItemReader"> 
     <property name="resources" value="#{jobParameters[INPUT_FILES]}" /> 
     <property name="delegate" ref="prT813InputReader" /> 
</bean> 

<bean id="prT813MultiThreadedReader" scope="step" class="org.springframework.batch.item.file.MultiResourceItemReader"> 
     <property name="resources" value="#{stepExecutionContext[fileName]}" /> 
     <property name="delegate" ref="prT813InputReader" /> 
    </bean> 

<bean id="prT813InputReader" scope="step" class="com.fileprocessing.ParticipantRecordT813ItemReader"> 
     <property name="delegate" ref="prT813CustomPeekableItemReader" /> 
    </bean> 

    <bean id="prT813CustomPeekableItemReader" scope="step" class="org.springframework.batch.item.support.SingleItemPeekableItemReader"> 
     <property name="delegate" ref="participantRecordT813ItemReader" /> 
    </bean> 

    <bean id="participantRecordT813ItemReader" scope="step" class="org.springframework.batch.item.file.FlatFileItemReader"> 
     <property name="lineMapper" ref="prT813LineMapper" /> 
    </bean> 


Created a new Reader class: 
public class ParticipantRecordT813ItemReader extends SingleItemPeekableItemReader<ParticipantRecordT813InputData> { 

    private static final String CLASS = "ParticipantRecordT813ItemReader"; 

    @Override 
    public ParticipantRecordT813InputData read() throws UnexpectedInputException, ParseException, Exception { 

     ParticipantRecordT813InputData item = super.read(); 
     Log.report(CLASS, "I am in the reader ::::"); 
     if (item != null) { 
      while (item.getDeNum()=="551") { 
       Log.report(CLASS, "I am in the reader at DE551::::" + item.getDeNum()); 
       ParticipantRecordT813InputData possibleRelatedObject = peek(); 
       if (possibleRelatedObject == null) { 
        return item; 
       } 

       //logic to determine if next line in file relates to same object 
       boolean matches = possibleRelatedObject.getDeNum()=="552"; 

       if (matches) { 
        Log.report(CLASS, "I am in the reader at DE552::::" + possibleRelatedObject.getDeNum()); 
       } else { 
        return item; 
       } 
      } 
     } 
     return item; 
    } 
} 

I am getting the below exception: 

ERROR [main] (AbstractStep.java:225)- Encountered an error executing step stepB in job prT813FileReaderJob 
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.prT813MultiReader' defined in URL []: Initialization of bean failed; nested exception is org.springframework.beans.ConversionNotSupportedException: Failed to convert property value of type 'com.sun.proxy.$Proxy10 implementing org.springframework.batch.item.ItemStreamReader,org.springframework.batch.item.PeekableItemReader,java.io.Serializable,org.springframework.aop.scope.ScopedObject,org.springframework.aop.framework.AopInfrastructureBean,org.springframework.aop.SpringProxy,org.springframework.aop.framework.Advised' to required type 'org.springframework.batch.item.file.ResourceAwareItemReaderItemStream' for property 'delegate'; nested exception is java.lang.IllegalStateException: Cannot convert value of type [com.sun.proxy.$Proxy10 implementing org.springframework.batch.item.ItemStreamReader,org.springframework.batch.item.PeekableItemReader,java.io.Serializable,org.springframework.aop.scope.ScopedObject,org.springframework.aop.framework.AopInfrastructureBean,org.springframework.aop.SpringProxy,org.springframework.aop.framework.Advised] to required type [org.springframework.batch.item.file.ResourceAwareItemReaderItemStream] for property 'delegate': no matching editors or conversion strategy found 







As you can see that prT813MultiReader and prT813MultiThreadedReader of type MultiResourceItemReader and I delegate them to prT813InputReader of type SingleItemPeekableItemReader. 







I tried implementing ResourceAwareItemReaderItemStream in my reader class which get rid of the above exception but then it complaints on ParticipantRecordT813InputData item = super.read(); for nullPointerException. 

public class ParticipantRecordT813ItemReader extends SingleItemPeekableItemReader<ParticipantRecordT813InputData> implements ResourceAwareItemReaderItemStream<ParticipantRecordT813InputData> { 

    private static final String CLASS = "ParticipantRecordT813ItemReader"; 
    SingleItemPeekableItemReader<ParticipantRecordT813InputData> delegate = new SingleItemPeekableItemReader<ParticipantRecordT813InputData>(); 

    @Override 
    public ParticipantRecordT813InputData read() throws UnexpectedInputException, ParseException, Exception { 

     ParticipantRecordT813InputData item = super.read(); 
     Log.report(CLASS, "I am in the reader ::::"); 
     if (item != null) { 
      while (item.getDeNum()=="551") { 
       Log.report(CLASS, "I am in the reader at DE551::::" + item.getDeNum()); 
       ParticipantRecordT813InputData possibleRelatedObject = peek(); 
       if (possibleRelatedObject == null) { 
        return item; 
       } 

       //logic to determine if next line in file relates to same object 
       boolean matches = possibleRelatedObject.getDeNum()=="552"; 

       if (matches) { 
        Log.report(CLASS, "I am in the reader at DE552::::" + possibleRelatedObject.getDeNum()); 
       } else { 
        return item; 
       } 
      } 
     } 
     return item; 
    } 

    @Override 
    public void close() throws ItemStreamException { 
     // TODO Auto-generated method stub 
     super.close(); 
    } 

    @Override 
    public void open(ExecutionContext arg0) throws ItemStreamException { 
     // TODO Auto-generated method stub 
     super.open(arg0); 
    } 

    @Override 
    public void update(ExecutionContext arg0) throws ItemStreamException { 
     // TODO Auto-generated method stub 
     super.update(arg0); 
    } 

    @Override 
    public void setResource(Resource arg0) { 
     // TODO Auto-generated method stub 
     super.setDelegate(delegate);   
    } 
} 


Any idea where I am wrong????