将CompletionPolicy扩展为从MultiResourceItemReader完成/提交每个资源

Question

我的问题是我不理解各种Spring批处理上下文。参考留档解释了如何将数据传递给未来的步骤。但是我如何在步骤内的读取器和写入器组件之间传递数据。步骤上下文。是否有块上下文？我以前在编写分区程序时在执行上下文之前使用过。但这些是并行执行的。

我现在需要做一个有序的手术。它基本上是一个jdbc导入作业，但每个文件都需要提交，否则它们就是外键约束。

我可以获得单个文件资源的行数的最简单的地方是在MultiResourceItemReader中，然后将其委托给ItemReader。但是在查看了各种CompletionPolicy实现之后，他们似乎只能访问RepeatContext。如何在MultiResourceItemReader的RepeatContext中存储一个值，以便我的CompletionPolicy可以访问它并在特定的文件资源行数之后提交。

举一个例子，说明如何扩展抽象计数完成策略，以及如何从MultiResourceItemReader存储数据。

或者也许有更好的方法来评估这类工作。

<!-- <job id="job" restartable="${restartable}" xmlns="http://www.springframework.org/schema/batch"> -->
    <batch:job id="job" restartable="true"
        xmlns="http://www.springframework.org/schema/batch">
        <batch:step id="step1-unzipFile">
            <batch:tasklet ref="unzipFileTasklet" />
            <batch:next on="COMPLETED" to="step2-import" />
        </batch:step>
        <batch:step id="step2-import"> <!-- we can't use a commit-interval="${commitInterval} cause it messes with 2nd pass import processing if the commit ends up being the middle of the file -->
            <batch:tasklet>
                <batch:chunk reader="multiResourceReader" writer="itemWriter" chunk-completion-policy="completionPolicy"/>  
            </batch:tasklet>
            <!-- <batch:next on="COMPLETED" to="step3-fileCleanUp" /> -->
        </batch:step>
        <!-- <batch:step id="step3-fileCleanUp">
            <batch:tasklet ref="fileCleanUpTasklet" />
        </batch:step> -->
    </batch:job>




    <bean id="multiResourceReader" class="springbatch.iimport.extended.SequentialLoaderMultiFileResourceItemReader" scope="step">
        <property name="resourceDirectoryPath" value="${importTempDirectoryBasePath}/#{jobParameters['jobKey']}/"/>
        <property name="delegate" ref="itemReader"/>
    </bean>

    <bean id="itemReader" class="org.springframework.batch.item.file.FlatFileItemReader">
        <property name="lineMapper" ref="lineMapper" />
    </bean>

    <bean id="lineMapper" class="springbatch.iimport.extended.JsonToTupleLineMapper"/>

    <bean id="itemWriter" class="springbatch.iimport.extended.TupleJdbcBatchItemWriter" scope="step">
        <property name="moduleDataSource" ref="moduleDataSource"/>
        <property name="dataSource" ref="dataSource"/>
        <property name="jobKey" value="#{jobParameters['jobKey']}"/>
        <property name="jobDef" value="#{jobParameters['jobDef']}"/>
    </bean>

    <bean id="completionPolicy" class="?"/> 

    <!-- tasklets -->

    <bean id="unzipFileTasklet" class="springbatch.iimport.tasklets.UnZipFile" scope="step">
        <!-- the temp directory the files are unzipped to end up being #{jobParameters['importZipFileName']} -->
        <property name="importZipFileName" value="${uploadDir}/#{jobParameters['importZipFileName']}" />
        <property name="jobKey" value="#{jobParameters['jobKey']}"/>
        <property name="importTempDirectoryBasePath" value="${importTempDirectoryBasePath}" />
     </bean>

Answer 1

您可以编写自己的CompletionPolicy，它前瞻itemReader（使用peakableitemreader）并在下次调用itemReader时将当前区块返回为“已完成”。next（）返回null（表示当前文件的EOF）
请记住：此解决方案可能会导致内存问题，因为内存中的文件内容已被完全读取。

public class EOFCompletionPolicy extends CompletionPolicySupport
{
    private EOFCompletionContext cc;
    private PeekableItemReader<Object> reader;

    public void setReader(PeekableItemReader<Object> forseeingReader)
    {
        this.reader = forseeingReader;
    }

    @Override
    public boolean isComplete(RepeatContext context)
    {
        return this.cc.isComplete();
    }

    @Override
    public RepeatContext start(RepeatContext context)
    {
        this.cc = new EOFCompletionContext(context);
        return cc;
    }

    @Override
    public void update(RepeatContext context)
    {
        this.cc.update();
    }

    protected class EOFCompletionContext extends RepeatContextSupport
    {
        boolean eof = false;
        public EOFCompletionContext (RepeatContext context)
        {
            super(context);
        }

        public void update()
        {
            final Object next;
            try
            {
                next = reader.peek();
            }
            catch (Exception e)
            {
                throw new NonTransientResourceException("Unable to peek", e);
            }
            // EOF?
            this.eof = (next == null);
        }

        public boolean isComplete() {
            return this.eof;
        }
    }
}