java - Spring Batch 划分一个步骤

Question

我有多个 CSV 文件要读取。我希望一次处理一个文件。而不是读取所有记录直到达到提交级别。

我已经整理了一个使用分区的作业，但在运行该作业时，我发现每行都有两个条目。就好像作业运行了两次一样。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
    xmlns:batch="http://www.springframework.org/schema/batch" xmlns:mvc="http://www.springframework.org/schema/mvc"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans  
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
http://www.springframework.org/schema/context  
http://www.springframework.org/schema/context/spring-context-4.0.xsd  
http://www.springframework.org/schema/mvc  
http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd  
http://www.springframework.org/schema/batch   
http://www.springframework.org/schema/batch/spring-batch-2.2.xsd">

    <import resource="classpath:/database.xml" />



     <bean id="asyncTaskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" >
     <property name="concurrencyLimit" value="1"></property>
     </bean>  

     <bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <property name="corePoolSize" value="5" />
    </bean>

     <bean id="partitioner" class="org.springframework.batch.core.partition.support.MultiResourcePartitioner" scope="step">
        <property name="resources" value="file:#{jobParameters[filePath]}/*.dat" />
    </bean>

    <bean id="multiResourceReader"
        class="org.springframework.batch.item.file.MultiResourceItemReader"
        scope="step">
        <property name="resources" value="file:#{jobParameters[filePath]}/*.dat"></property>
        <property name="delegate" ref="logItFileReader"></property>
    </bean>



    <batch:job id="remediationJob">
        <batch:step id="partitionedStep" >
            <batch:partition step="readWriteContactsPartitionedStep" partitioner="partitioner">
                <batch:handler task-executor="asyncTaskExecutor" />
            </batch:partition>
        </batch:step>
    </batch:job>

    <batch:step id="readWriteContactsPartitionedStep">
        <batch:tasklet>
            <batch:transaction-attributes isolation="READ_UNCOMMITTED"/>
            <batch:chunk reader="multiResourceReader" writer="rawItemDatabaseWriter" commit-interval="10" skip-policy="pdwUploadSkipPolicy"/>
        <batch:listeners>
                    <batch:listener ref="customItemReaderListener"></batch:listener>
                    <batch:listener ref="csvLineSkipListener"></batch:listener>
                    <batch:listener ref="getCurrentResourceChunkListener"></batch:listener>

                </batch:listeners>
        </batch:tasklet>    
    </batch:step>


    <bean id="logItFileReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="step">
        <!-- Read a csv file -->

        <property name="strict" value="false"></property>
        <property name="lineMapper">
            <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
                <!-- split it -->
                <property name="lineTokenizer">
                    <bean
                        class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
                        <property name="delimiter" value="@##@" />
                        <property name="strict" value="true" />
                    </bean>
                </property>
                <property name="fieldSetMapper">
                    <!-- map to an object -->
                    <bean class="org.kp.oppr.remediation.batch.vo.CSVDataVOFieldMapper">

                    </bean>
                </property>
            </bean>
        </property>
    </bean>

    <bean id="rawItemDatabaseWriter" class="org.kp.oppr.remediation.batch.csv.RawItemDatabaseWriter"
        scope="step">
    </bean>

    <bean id="pdwUploadSkipPolicy"
        class="org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy" />

    <bean id="csvDataVO" class="org.kp.oppr.remediation.batch.vo.CSVDataVO"
        scope="prototype"></bean>


    <!-- BATCH LISTENERS -->

    <bean id="pdwFileMoverListener"
        class="org.kp.oppr.remediation.batch.listener.PdwFileMoverListener"
        scope="step">
    </bean>

    <bean id="csvLineSkipListener"
        class="org.kp.oppr.remediation.batch.listener.CSVLineSkipListener"
        scope="step">
    </bean>

    <bean id="customItemReaderListener"
        class="org.kp.oppr.remediation.batch.listener.CustomItemReaderListener"></bean>

     <bean id="getCurrentResourceChunkListener" 
          class="org.kp.oppr.remediation.batch.listener.GetCurrentResourceChunkListener">
        <property name="proxy" ref ="multiResourceReader" />
    </bean>
    <!-- 
    <bean id="stepListener" class="org.kp.oppr.remediation.batch.listener.ExampleStepExecutionListener">
        <property name="resources" ref="multiResourceReader"/>
    </bean>
     -->
    <!-- Skip Policies -->

</beans>  

我在这里缺少什么吗？

Answer 1

您有 2 个问题

1 - “我希望一次处理一个文件。而不是读取所有记录直到达到提交级别。” 将 Commit-Interval 设置为 1 - 它将读取一个项目，对其进行处理，然后编写器将等待，直到有 1 个项目要写入。

2 - 就好像作业运行了两次。

看起来它会运行与您拥有的文件数量一样多的次数。

此步骤不应使用 MultiResourceItemReader。 分区器将资源分割成多个并创建单独的执行上下文。由于设置了资源属性，您的 MultiResourceItemReader 再次考虑所有文件。