java - 如何在 Spring Batch : Updated 中循环运行步骤

Question

我有以下 Spring 批量作业:

<job id="MyBatchJob" job-repository="jobRepository">

  <step id="ConfigurationReadStep">
        <tasklet ref="ConfigurationReadTasklet"  transaction-manager="jobRepository-transactionManager"/>
        <next on="COMPLETED"  to="NextStep" />  
  </step>        

  <step id="NextStep">
    <tasklet transaction-manager="jobRepository-transactionManager">
        <chunk reader="myItemReader" writer="myItemWriter" commit-interval="1000"/>
    </tasklet>
  </step>

     <listeners>
        <listener ref="jobListener" />
     </listeners>     

</job>

它的第一步是配置读取步骤，在一些业务逻辑之后，我遇到了一个查询列表。例如说出 10 个查询。我知道我可以使用 JobExecutionContext 和 PromotionListener 来提升此列表。

但是，我想将此查询作为读取器查询一一提供给下一步的读取器，并循环运行该步骤，直到消耗完所有读取器查询。我想将每个查询作为 Spring 批处理场景运行，因为它们可能返回大量项目。

我该怎么做？

************************************ 更新************ **********************

这就是我正在尝试做的事情:

<job id="MyJob" job-repository="jobRepository">

      <step id="ConfigurationReadStep">
            <tasklet ref="ConfigurationReadTasklet"  transaction-manager="jobRepository-transactionManager"/>
            <next on="COMPLETED"  to="MyNextStep" />    
             <listeners>
                <listener ref="promotionListener"/>
            </listeners>
      </step>        

    <step id="MyNextStep" next="limitDecision">
        <tasklet transaction-manager="jobRepository-transactionManager">
            <chunk reader="MyItemReader" writer="MyItemWriter" commit-interval="1000"/>
        </tasklet>
    </step>

    <decision id="limitDecision" decider="limitDecider">
        <next on="CONTINUE" to="MyNextStep" />
        <end on="COMPLETED" />
    </decision>

     <listeners>
        <listener ref="jobListener" />
     </listeners>

  </job>

<beans:bean id="jobListener" class="com.hsbc.gbm.dml.integration.batch.listener.SimpleJobListener" />

<beans:bean id="promotionListener" class="org.springframework.batch.core.listener.ExecutionContextPromotionListener">
    <beans:property name="keys" value="readerLimit,readerQueries,readerSQL" />
</beans:bean>

  <beans:bean id="ConfigurationReadTasklet" class="com.mypackage.ConfigurationReadTasklet" scope="step">
  </beans:bean>

<beans:bean id="MyItemReader" class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step">
    <beans:property name="dataSource" ref="myDataSource" />
    <beans:property name="sql" value="#{jobExecutionContext['readerSQL']}" />
    <beans:property name="rowMapper">
        <beans:bean  name="mapper" class="com.mypackage.MyRowMapper" />
    </beans:property>
</beans:bean>

<beans:bean id="MyItemWriter" class="com.mypackage.MyItemWriter" scope="step">
    <beans:constructor-arg ref="myDataSource" />
</beans:bean>

<beans:bean id="limitDecider" class="com.mypackage.StepLoopLimitDecider" scope="step"> 
    <beans:property name="readerQueries" value="#{jobExecutionContext['readerQueries']}" />
</beans:bean>     

在 tasklet 中，我将获取读者查询列表。我将把 readerQuery 设置为其中的第一个，然后进入下一步，该步骤将作为 Spring Batch 步骤正常运行。

完成后，我希望我的 Decider 检查是否有更多读者查询，如果有，它将把 readerQuery 更改为下一个查询并重新运行 NextStep，否则作业将完成。以下是我的决策者:

public class StepLoopLimitDecider implements JobExecutionDecider {

private List<String> readerQueries;

private int count = 1;

public FlowExecutionStatus decide(JobExecution jobExecution,StepExecution stepExecution) {
    if (count >= this.readerQueries.size()) {
        return new FlowExecutionStatus("COMPLETED");
    }
    else {
        System.out.println(count + ": "+readerQueries.get(count));
        jobExecution.getExecutionContext().put("readerSQL", readerQueries.get(count));
        count = count + 1;
        return new FlowExecutionStatus("CONTINUE");
    }
}

public void setReaderQueries(List<String> readerQueries) {
    this.readerQueries = readerQueries;
}   
}

但是，这不起作用。步骤第一次运行正确。但决策者失败并出现以下错误:

    2016-12-06 14:46:24 ERROR AbstractJob:306 - Encountered fatal error executing job
org.springframework.batch.core.JobExecutionException: Flow execution ended unexpectedly
    at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:141)
    at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:281)
    at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:120)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:48)
    at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:114)
    at org.springframework.batch.core.launch.support.CommandLineJobRunner.start(CommandLineJobRunner.java:348)
    at org.springframework.batch.core.launch.support.CommandLineJobRunner.main(CommandLineJobRunner.java:565)
Caused by: 
org.springframework.batch.core.job.flow.FlowExecutionException: Ended flow=MyBatchJob at state=MyBatchJob.limitDecision with exception
    at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:152)
    at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:124)
    at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:135)
    ... 6 more
Caused by: 
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.limitDecider': Scope 'step' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton; nested exception is java.lang.IllegalStateException: No context holder available for step scope
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:339)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:190)
    at org.springframework.aop.target.SimpleBeanTargetSource.getTarget(SimpleBeanTargetSource.java:33)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:182)
    at $Proxy4.decide(Unknown Source)
    at org.springframework.batch.core.job.flow.support.state.DecisionState.handle(DecisionState.java:43)
    at org.springframework.batch.core.configuration.xml.SimpleFlowFactoryBean$DelegateState.handle(SimpleFlowFactoryBean.java:141)
    at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:144)
    ... 8 more
Caused by: 
java.lang.IllegalStateException: No context holder available for step scope
    at org.springframework.batch.core.scope.StepScope.getContext(StepScope.java:197)
    at org.springframework.batch.core.scope.StepScope.get(StepScope.java:139)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:325)
    ... 15 more

有人可以帮我完成这项工作吗？

Answer 1

我建议您根据计算查询的业务逻辑结果动态构建作业。

但是，您无法使用 xml-configuration 执行此操作，因此，我建议您使用 java-config api。

我已经写了几个关于“如何动态创建工作”的类似问题的答案。如果您需要进一步的建议，请查看它们并告诉我。

SpringBatch - javaconfig vs xml

Spring Batch Java Config: Skip step when exception and go to next steps

Spring batch repeat step ending up in never ending loop

Spring Batch - How to generate parallel steps based on params created in a previous step

Spring Batch - Looping a reader/processor/writer step