java - Spring Batch和JobInstanceAlreadyCompleteException:

Question

我有一个Spring Batch作业，该作业通过SFTP从远程Linux服务器检索文件。
远程服务器上的目录是一个包含7天文件（约400个文件）的档案。文件的大小相对较小。

Spring Batch知道哪些文件已经被处理。

当我启动应用程序时。第一次，Spring Batch Tasklet检索文件，然后，Spring Batch为它已经处理的每个文件生成一个异常：

例如。

由以下原因引起：org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException：一个作业实例已经存在，并且对于parameter = {input.file.url = file：///blahblahblah.txt}而言是完整的。

这导致处理的巨大延迟。

第一次之后，在随后的sftp检索文件时，没有异常，因此也没有延迟（也许Spring Batch生成了已处理文件的内部哈希列表）。


在Transformer类中，是否应该检查文件是否在本地存在，并且仅在尚未处理的新文件上调用JobLaunchRequest（）？

/ **


将BAI文件转换为Spring Batch作业启动请求
* /
公共类FileMessageToJobRequestTransformer {
公共静态最终Logger LOGGER = LoggerFactory.getLogger（FileMessageToJobRequestTransformer.class）;
私人工作职位；

私有字符串fileParameterName;

public void setJob（工作职位）{
    this.job =工作；
}

公共无效setFileParameterName（String fileParameterName）{
    LOGGER.debug（“ file parameter name：{}”，fileParameterName）;
    this.fileParameterName = fileParameterName;
}

@变压器
公共JobLaunchRequest转换（消息）{
    LOGGER.debug（“文件消息：{}”，消息）；
    JobParametersBuilder jobParametersBuilder = new JobParametersBuilder（）;

jobParametersBuilder.addString(fileParameterName,
        "file://" + message.getPayload().getAbsolutePath());

LOGGER.debug("Job params: {}", jobParametersBuilder.toJobParameters());

return new JobLaunchRequest(job, jobParametersBuilder.toJobParameters());

}
}

有什么办法可以捕捉到异常吗？


JobInstanceAlreadyCompleteException


是否应该在SftpBaiParserJobBridge-context.hml中设置retry-limit =“ 1”和skip-limit =“ 1”？



<!-- When getting a new BAI file, transform into spring batch job request -->
<int:transformer id="fileMessageToJobRequestTransformer"
input-channel="inboundFileChannel"
output-channel="outboundJobRequestChannel"
method="transform">
<bean class="com.distributedfinance.mbi.bai.transformer.FileMessageToJobRequestTransformer">
<property name="job" ref="baiParseJob"/>
<property name="fileParameterName" value="input.file.url"/>
</bean>
<int:poller fixed-rate="10000"/>
</int:transformer>


附加addDate（）允许多次处理文件。
数据库中现在有dups。

jobParametersBuilder.addString(fileParameterName,
                "file://" + message.getPayload().getAbsolutePath())).addDate("rundate", new Date()).toJobParameters();

谢谢！


我可以通过'@Transformer内部的JobExplorer接口查询存储库来检查文件是否已被处理
public JobLaunchRequest transform（Message message）'，如果文件已经处理过，则返回null？


例如。

public interface JobExplorer {

    List<JobInstance> getJobInstances(String jobName, int start, int count);

    JobExecution getJobExecution(Long executionId);

    StepExecution getStepExecution(Long jobExecutionId, Long stepExecutionId);

    JobInstance getJobInstance(Long instanceId);

    List<JobExecution> getJobExecutions(JobInstance jobInstance);

    Set<JobExecution> findRunningJobExecutions(String jobName);
}

如何捕获异常？

由以下原因引起：org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException：一个作业实例已经存在，并且对于参数= {input.file.url = file：/// home / dlaxer / dfc-bank-integration / mbi- application / bai / download / BAI_Intraday160302070054471.txt}。如果要再次运行此作业，请更改参数。
        在org.springframework.batch.core.repository.support.SimpleJobRepository.createJobExecution（SimpleJobRepository.java:126）
        在sun.reflect.NativeMethodAccessorImpl.invoke0（本机方法）处
        在sun.reflect.NativeMethodAccessorImpl.invoke（NativeMethodAccessorImpl.java:62）
        在sun.reflect.DelegatingMethodAccessorImpl.invoke（DelegatingMethodAccessorImpl.java:43）
        在java.lang.reflect.Method.invoke（Method.java:498）
        在org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection（AopUtils.java:302）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint（ReflectiveMethodInvocation.java:190）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.proceed（ReflectiveMethodInvocation.java:157）
        在org.springframework.transaction.interceptor.TransactionInterceptor $ 1.proceedWithInvocation（TransactionInterceptor.java:99）
        在org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction（TransactionAspectSupport.java:281）
        在org.springframework.transaction.interceptor.TransactionInterceptor.invoke（TransactionInterceptor.java:96）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.proceed（ReflectiveMethodInvocation.java:179）
        在org.springframework.batch.core.repository.support.AbstractJobRepositoryFactoryBean $ 1.invoke（AbstractJobRepositoryFactoryBean.java:172）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.proceed（ReflectiveMethodInvocation.java:179）
        在org.springframework.aop.framework.JdkDynamicAopProxy.invoke（JdkDynamicAopProxy.java:207）
        在com.sun.proxy。$ Proxy113.createJobExecution（未知来源）
        在org.springframework.batch.core.launch.support.SimpleJobLauncher.run（SimpleJobLauncher.java:125）
        在sun.reflect.NativeMethodAccessorImpl.invoke0（本机方法）处
        在sun.reflect.NativeMethodAccessorImpl.invoke（NativeMethodAccessorImpl.java:62）
        在sun.reflect.DelegatingMethodAccessorImpl.invoke（DelegatingMethodAccessorImpl.java:43）
        在java.lang.reflect.Method.invoke（Method.java:498）
        在org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection（AopUtils.java:302）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint（ReflectiveMethodInvocation.java:190）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.proceed（ReflectiveMethodInvocation.java:157）
        在org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration $ PassthruAdvice.invoke（SimpleBatchConfiguration.java:127）
        在org.springframework.aop.framework.ReflectiveMethodInvocation.proceed（ReflectiveMethodInvocation.java:179）
        在org.springframework.aop.framework.JdkDynamicAopProxy.invoke（JdkDynamicAopProxy.java:207）
        在com.sun.proxy。$ Proxy114.run（未知来源）
        在org.springframework.batch.integration.launch.JobLaunchingMessageHandler.launch（JobLaunchingMessageHandler.java:50）处
        在org.springframework.batch.integration.launch.JobLaunchingGateway.handleRequestMessage（JobLaunchingGateway.java:76）
        ...另外35个

Answer 1

我看到您在工作启动器中仅传递了一个参数作为“文件名”。
作业启动器启动后，它将从存储库BATCH_JOB_EXECUTION表中进行查询，并将检查最后处理的作业的状态。
在当前作业执行中，输入参数与先前执行的作业相同，并且批处理状态和退出代码=已完成，那么您将获得JobInstanceAlreadyCompleteException。您应该尝试在每次执行时传递始终唯一的Parmae​​ter。只需传递当前时间作为参数并尝试

JobParameters jobparam =   new JobParametersBuilder().addString(fileParameterName, "file://" + message.getPayload().getAbsolutePath())
                        .addDate("rundate", new Date()).toJobParameters();

JobExecution execution = jobLauncher.run(job, jobparam);
         catch (Exception e) {
             if ( e instanceof JobInstanceAlreadyCompleteException){
                 System.out.println("Raj*************");
             }

2是的，您可以像这样处理

catch（例外e）{
             if（eInstanceof JobInstanceAlreadyCompleteException）{
                 System.out.println（“需要处理*************”）；
             }