Spring Batch - 从另一个具有重试机制的Tasklet调用Tasklet?

Spring Batch 我期待实现重试机制 . 我有下面配置的步骤

此批作业实际运行并创建付款CSV文件,并将它放置在我的应用程序服务器的出站文件夹,并通过微进程,我呼吁其拾取的所有文件,并将数据发送到另一台服务器的shell脚本 . 在通过批处理作业执行shell脚本时,某些操作失败,那么我需要实现重试机制 .

如何使用重试机制从另一个Tasklet调用Tasklet?

public class ABCTasklet implements Tasklet, InitializingBean {

    @Value("${payment.directory}") // location where files are written - CSV
    private String location;

    @Override
    public RepeatStatus execute(StepContribution step, ChunkContext chunkContext) throws Exception {
        Set<String> payFileNames = new HashSet<>();
        List<FileDetails> fileStatusList = new ArrayList<>();

        File folder = new File(location);

        for (File file : folder.listFiles()) {
            FileDetails fileDetails = new FileDetails();
            fileDetails.setFileName(file.getName());
            fileDetails.setStatus(FileStatus.FAILURE.getStatus());
            fileStatusList.add(fileDetails);
            payFileNames.add(file.getName());
        }
        chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext().put("payFileNames", payFileNames);
        chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext().put("fileStatusList", fileStatusList);
        return RepeatStatus.FINISHED;
    }

    public String getLocation() {
        return location;
    }

    public void setLocation(String location) {
        this.location = location;
    }
}

Another Tasklet:

public class XYZListener implements Tasklet {
    @Autowired
    private RetryTemplate retryTemplate;

    private ExecutionContext executionContext;
    public ExecutionContext getExecutionContext() {
        return executionContext;
    }
    public void setExecutionContext(ExecutionContext executionContext) {
        this.executionContext = executionContext;
    }

    @Override
    public RepeatStatus execute(StepContribution paramStepContribution,ChunkContext paramChunkContext) throws Exception {
        Set<String> paymentFileNames = new HashSet<>();
        if(null != executionContext && null != executionContext.get("paymentFileNames")) {
            paymentFileNames = (Set<String>) executionContext.get("paymentFileNames");
        }

        try {
            for(String payFileName : paymentFileNames) {
                int exitStatus = 1;
                if (payFileName != null) {
    // execute shell scripts
                    ProcessBuilder procBuilder = new ProcessBuilder("/bin/bash","paymentBatchFiles.sh", payFileName);

                    Map<String, String> env = procBuilder.environment();
                    env.put("PATH", "/usr/local/bin:/bin:XXXXXXXXXXXXXXXXXX"); // many linux env. variables
                    env.put("MAYMENT_HOME","/Deployment/PaymentData"); 
                    ....
                    .....// other needed env variables

                    Process proc = procBuilder.start();
                    BufferedReader br = new BufferedReader(new InputStreamReader(proc.getErrorStream()));
                    exitStatus = proc.waitFor();
                }

                if (0 == exitStatus) {
                    paymentDetailsDAO.updateStatus(EDSProcessStatus.SENT.getStatusCode(), payFileName, JOB_NAME);
                }
                else {
                     // Implement Retry mechanism to call the ABCTasklet
                    retryTemplate.execute(); //TODO                  
                }
            }
            return RepeatStatus.FINISHED;
        } catch (InterruptedException | IOException ie) {

        }
    }
}

工作细节

<batch:decision id="paymentDecision" decider="paymentDecider">
    <batch:next on="FAILURE" to="failureStep" />
    <batch:next on="SUCCESS" to="x" />
</batch:decision>

<batch:step id="x">
    <batch:tasklet ref="payTasklet" />
    <batch:next on="COMPLETED" to="step1" />
</batch:step>

<batch:step id="step1">
    <batch:tasklet>
        <batch:chunk reader="paymentReader" writer="paymentWriter" commit-interval="100">
            <batch:listeners>
                <batch:listener ref="XYZListener" />
            </batch:listeners>
        </batch:chunk>
    </batch:tasklet>
    <batch:next on="COMPLETED" to="sendPaymentBatchFiles" />
</batch:step>

<batch:step id="sendPaymentBatchFiles">
    <batch:tasklet ref="sendPaymentBatchFilesTasklet">
    </batch:tasklet>
    <batch:next on="COMPLETED" to="lastRunUpdate" />
</batch:step>

<batch:step id="lastRunUpdate" next="Z">
    <batch:tasklet ref="QQQQRunUpdater" />
</batch:step>


<batch:step id="Z">
    <batch:tasklet ref="PPTasklet" />
</batch:step>

<batch:step id="failureStep" next="sendPaymentBatchFiles">
    <batch:tasklet ref="ABCTasklet" />
</batch:step>

如何从XYZListener调用重试机制即ABCTasklet类?代码更新将不胜感激

EDIT-1 我在下面做了,但似乎没有工作?

return retryTemplate.execute(new RetryCallback<RepeatStatus, RuntimeException>() {
    @Override
    public RepeatStatus doWithRetry(RetryContext context) {
        logger.info("Retry File, Retry Count = "+context.getRetryCount());
        return executeShellScript();
    }
});

Edit-2:

RepeatTemplate template = new RepeatTemplate();
template.setCompletionPolicy(new DefaultResultCompletionPolicy());

StringRepeatCallback callback = new StringRepeatCallback();
callback.setBillingFileNames(billingFileNames);
template.iterate(callback);

另一课:

public class StringRepeatCallback implements RepeatCallback{
    @Autowired
    private PaymentFilesTasklet tasklet;

    private Set<String> paymentFileNames;

    public void setBillingFileNames(Set<String> paymentFileNames) {
        this.paymentFileNames = paymentFileNames;
    }

    @Override
    public RepeatStatus doInIteration(RepeatContext repeatContext) throws Exception {
        tasklet.executeShellScript(paymentFileNames);
        return RepeatStatus.CONTINUABLE;
    }
}

我得到的错误是:

2018-07-22 13:12:28,287 ERROR [main] (AbstractStep.java:229) - Encountered an error executing step sendPaymentBatchFiles in job paymentBatchJob
java.lang.NullPointerException
        at com.mastercard.ess.eds.billing.callback.StringRepeatCallback.doInIteration(StringRepeatCallback.java:24)
        at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
        at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
        at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
        at com.mastercard.ess.eds.billing.tasklet.SendBillingBatchFilesTasklet.executeShellScript(SendBillingBatchFilesTasklet.java:132)
        at com.mastercard.ess.eds.billing.tasklet.SendBillingBatchFilesTasklet.execute(SendBillingBatchFilesTasklet.java:152)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
        at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:133)
        at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:121)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
        at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
        at com.sun.proxy.$Proxy44.execute(Unknown Source)
        at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:406)
        at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:330)
        at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133)
        at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271)
        at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:81)
        at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
        at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
        at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
        at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
        at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200)
        at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
        at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:64)
        at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:67)
        at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:169)
        at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:144)
        at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:134)
        at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:306)
        at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:135)
        at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
        at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:128)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
        at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
        at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
        at com.sun.proxy.$Proxy34.run(Unknown Source)
        at org.springframework.batch.core.launch.support.CommandLineJobRunner.start(CommandLineJobRunner.java:362)
        at org.springframework.batch.core.launch.support.CommandLineJobRunner.main(CommandLineJobRunner.java:592)
2018-07-22 13:12:28,459 ERROR [main] (AbstractJob.java:335) - Encountered fatal error executing job

回答(0)