批处理作业失败后如何停止或暂停轮询？答案

【问题标题】：How to stop or suspend polling after batch job fail?批处理作业失败后如何停止或暂停轮询？
【发布时间】：2018-05-19 19:00:26
【问题描述】：

我们使用 spring-batch-integration 来处理目录中的 .json 文件。必须在失败后停止处理，找出问题（更换有问题的文件或其他解决方案）然后继续。当前配置在出错后继续轮询。如何改变它？或者对于这种情况可能有不同的方法。

@Configuration
@IntegrationComponentScan
@EnableIntegration
public class IntegrationConfig {

private @Autowired Job job;

@Bean
@ServiceActivator(inputChannel = "jobChannel", 
   outputChannel = "errorChannel")
protected JobLaunchingMessageHandler launcher(JobLauncher jobLauncher) {
    return new JobLaunchingMessageHandler(jobLauncher);
}

@Bean
public MessageChannel fileInputChannel() {
    return new DirectChannel();
}

@Bean
@InboundChannelAdapter(value = "fileInputChannel",
                       poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> fileReadingMessageSource() {
    FileReadingMessageSource source = new FileReadingMessageSource();
    source.setDirectory(new File("C:/Temp/myfiles/"));
    source.setFilter(new SimplePatternFileListFilter("*.json"));
    source.setScanEachPoll(true);
    source.setUseWatchService(true);
    return source;
}

@Transformer(inputChannel = "fileInputChannel",
             outputChannel = "jobChannel")
public JobLaunchRequest transform(File aFile) {
    String fileName = aFile.getAbsolutePath();
    JobParameters jobParameters =
    new JobParametersBuilder().addString("input.file.name", fileName)
            .addDate("dateTime", new Date()).toJobParameters();
    JobLaunchRequest request = new JobLaunchRequest(job, jobParameters);
    return request;
}
}

example was from this article

outputChannel = "nullChannel" 和 outputChannel = "errorChannel" 都没有帮助

【问题讨论】：

标签： spring-integration spring-batch

【解决方案1】：

您需要停止入站通道适配器。

您可以自动装配一个由@InboundChannelAdapter 注解注册的SourcePollingChannelAdapter。

当您检测到故障时，在适配器上调用stop()。

【讨论】：

我遇到了自动装配错误，请看我的回答，自动装件的正确方法是什么？
您不应添加答案以显示更多详细信息；你应该编辑你的问题。尝试将适配器设为静态：public static MessageSource<File> fileReadingMessageSource() {，以便更早注册。

【解决方案2】：

我加了

@Bean
@DependsOn("fileInputChannel")
@ServiceActivator(inputChannel = "errorChannel", 
  outputChannel = "nullChanel")
protected ErrorLogger errorLogger(JobLauncher jobLauncher) {
    return new ErrorLogger();
}

和

public class ErrorLogger {
private static final Logger logger = 
LoggerFactory.getLogger(ErrorLogger.class);

@Autowired
private SourcePollingChannelAdapter fileInputChannel;


@ServiceActivator
public void logError(Message<JobExecution> message) {
    JobExecution msgex=message.getPayload();
     if (msgex.getStatus() == BatchStatus.FAILED) {
         logger.error("Exception " + 
         msgex.getExitStatus().getExitDescription());
         fileInputChannel.stop();
     }
}
}

但我在 ErrorLogger 中遇到自动装配错误

Unsatisfied dependency expressed through field 'fileInputChannel'; nested 
exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: 
No qualifying bean of type 
'org.springframework.integration.endpoint.SourcePollingChannelAdapter' 
available:

尽管有 @DependsOn("fileInputChannel")，但似乎是初始化顺序问题，因为我可以在单独的控制器中自动处理它而不会出错。

它只适用于

 @Autowired(required = false)
 private SourcePollingChannelAdapter fileInputChannel;

【讨论】：