Spring Integration轮询器遇JobLaunchingGateway异常中断当前轮询的解决建议

针对你遇到的单文件处理异常导致整轮批次后续文件终止以及应用重启后重复处理已完成文件的问题，以下是几个实用的优化方案：

方案一：在集成流中添加异常隔离处理 #

通过Spring Integration的请求处理器通知（Request Handler Advice），捕获单个文件处理时的异常，将错误文件移至错误目录并吞掉异常，确保当前轮询批次的其他文件能继续处理。

实现代码： #

@Bean
public ExpressionEvaluatingRequestHandlerAdvice fileProcessingErrorAdvice(FileProperties properties) {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    
    // 异常发生时执行：将文件移动到错误目录
    advice.setOnFailureExpressionString(
        "T(java.nio.file.Files).move(" +
        "payload.jobParameters.get('inputFile').toPath(), " +
        "new java.io.File('" + properties.getErrorDir() + "').toPath().resolve(payload.jobParameters.get('inputFile').getName()), " +
        "java.nio.file.StandardCopyOption.REPLACE_EXISTING)"
    );
    
    advice.setTrapException(true); // 吞掉异常，不中断后续消息处理
    advice.setFailureChannelName("fileProcessingErrorChannel"); // 可选：将异常消息发送到错误通道做日志记录
    
    return advice;
}

// 在集成流中为JobLaunchingGateway配置该通知
@Bean
public IntegrationFlow myIntegrationFlow(JobLaunchingGateway jobLaunchingGateway,
                                         FileMessageToJobRequest fileMessageToJobRequest,
                                         ExpressionEvaluatingRequestHandlerAdvice fileProcessingErrorAdvice) {
    return IntegrationFlows.from(Files.inboundAdapter(new File(properties.getInputDir()))
                    .filter(new AcceptOnceFileListFilter<>()),
            c -> c.poller(Pollers.fixedRate(300, TimeUnit.SECONDS)
                            .taskExecutor(taskExecutor())
                            .maxMessagesPerPoll(50)
                    ))
            .transform(fileMessageToJobRequest)
            .handle(jobLaunchingGateway, e -> e.advice(fileProcessingErrorAdvice)) // 绑定异常处理通知
            .log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload")
            .get();
}

// 可选：错误通道的日志处理
@Bean
public IntegrationFlow errorHandlingFlow() {
    return IntegrationFlows.from("fileProcessingErrorChannel")
            .log(LoggingHandler.Level.ERROR, "Error processing file: " + "#{payload.cause.message}")
            .get();
}

方案二：替换为持久化文件过滤器（从源头避免重复读取） #

将内存级的AcceptOnceFileListFilter替换为持久化缓存的过滤器，确保应用重启后仍能识别已处理过的文件，从根源上避免重复触发JobInstanceAlreadyCompleteException。

实现代码： #

@Bean
public MetadataStore persistentMetadataStore(DataSource dataSource) {
    // 使用JDBC持久化缓存，也可替换为RedisMetadataStore等其他实现
    return new JdbcMetadataStore(dataSource);
}

@Bean
public FileListFilter<File> persistentFileFilter(MetadataStore persistentMetadataStore) {
    MetadataStoreAwareFileListFilter<File> filter = new MetadataStoreAwareFileListFilter<>();
    filter.setMetadataStore(persistentMetadataStore);
    filter.setFilter(new AcceptOnceFileListFilter<>()); // 基于持久化缓存实现去重
    return filter;
}

// 更新集成流的文件适配器过滤器
@Bean
public IntegrationFlow myIntegrationFlow(JobLaunchingGateway jobLaunchingGateway,
                                         FileMessageToJobRequest fileMessageToJobRequest,
                                         FileListFilter<File> persistentFileFilter) {
    return IntegrationFlows.from(Files.inboundAdapter(new File(properties.getInputDir()))
                    .filter(persistentFileFilter), // 使用持久化过滤器
            c -> c.poller(Pollers.fixedRate(300, TimeUnit.SECONDS)
                            .taskExecutor(taskExecutor())
                            .maxMessagesPerPoll(50)
                    ))
            .transform(fileMessageToJobRequest)
            .handle(jobLaunchingGateway)
            .log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload")
            .get();
}

方案三：前置检查Job实例状态 #

在启动Job前，先通过JobRepository查询该文件对应的Job实例是否已完成，若已完成则直接将文件移至已处理目录，跳过Job启动流程。

实现代码： #

@Bean
public MessageHandler jobPreCheckHandler(JobLaunchingGateway jobLaunchingGateway,
                                         JobRepository jobRepository,
                                         FileProperties properties) {
    return message -> {
        JobRequest jobRequest = (JobRequest) message.getPayload();
        JobParameters jobParameters = jobRequest.getJobParameters();
        File inputFile = jobParameters.getFile("inputFile");
        
        try {
            // 查询对应Job实例的最后执行状态
            JobInstance jobInstance = jobRepository.createJobInstance(jobRequest.getJob().getName(), jobParameters);
            JobExecution lastExecution = jobRepository.getLastJobExecution(jobInstance);
            
            if (lastExecution != null && BatchStatus.COMPLETED.equals(lastExecution.getStatus())) {
                // 已完成，直接移至已处理目录
                Files.move(inputFile.toPath(), 
                        new File(properties.getProcessedDir()).toPath().resolve(inputFile.getName()),
                        StandardCopyOption.REPLACE_EXISTING);
                return;
            }
        } catch (NoSuchJobInstanceException e) {
            // 无该Job实例，正常执行后续流程
        } catch (IOException e) {
            throw new RuntimeException("Failed to move processed file", e);
        }
        
        // 正常启动Job
        jobLaunchingGateway.handleMessage(message);
    };
}

// 更新集成流，添加前置检查
@Bean
public IntegrationFlow myIntegrationFlow(MessageHandler jobPreCheckHandler,
                                         FileMessageToJobRequest fileMessageToJobRequest) {
    return IntegrationFlows.from(Files.inboundAdapter(new File(properties.getInputDir()))
                    .filter(new AcceptOnceFileListFilter<>()),
            c -> c.poller(Pollers.fixedRate(300, TimeUnit.SECONDS)
                            .taskExecutor(taskExecutor())
                            .maxMessagesPerPoll(50)
                    ))
            .transform(fileMessageToJobRequest)
            .handle(jobPreCheckHandler) // 前置检查Job状态
            .log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload")
            .get();
}

方案选择建议 #

1. 优先选方案二：从源头避免重复读取已处理文件，彻底解决JobInstanceAlreadyCompleteException问题，同时不影响原有错误处理逻辑。
2. 若无法使用持久化缓存：选择方案一，快速实现异常隔离，确保单个文件错误不影响整批处理。
3. 需要精细控制Job执行逻辑：选择方案三，在Job启动前做状态校验，更灵活地处理重复文件场景。

内容的提问来源于stack exchange，提问作者tardistraveller