Spring Integration Flow中SFTP出站适配器失败后仍执行后续订阅者问题

看起来你遇到的核心问题是PublishSubscribeChannel的默认行为导致订阅者独立执行，即使第一个SFTP上传订阅者失败，后面的归档和日志订阅者还是会继续运行，进而产生误报日志，同时deleteFileAdvice因为上传失败也不会触发。我来帮你梳理下原因和解决方案：

问题根源分析 #

1. PublishSubscribeChannel默认无异常传播机制：默认情况下，PublishSubscribeChannel会让每个订阅者独立执行（即使同步模式下，单个订阅者的异常也不会阻断其他订阅者）。如果你给SFTP出站适配器单独配置了errorChannel，异常会被该局部通道吞掉，主流程完全感知不到失败，自然会继续执行后续订阅者。
2. 业务逻辑顺序不合理：当前你把上传、归档、日志放在同一级分发，但实际业务逻辑应该是上传成功后才执行归档和日志，而不是不管上传结果都执行。

解决方案 #

方案1：调整流程顺序，上传成功后再触发后续操作 #

最贴合业务逻辑的方式是先确保SFTP上传成功，再分发到归档和日志流程，通过successChannel实现这一触发逻辑：

@Bean
public IntegrationFlow mainSftpFlow() {
    return IntegrationFlows.from(sftpInboundAdapter(), 
                    e -> e.poller(pollerConfig())) // 你的轮询器配置（Cron、错误通道等）
            .transform(fileTransformer()) // 你的文件转换逻辑
            // 先处理SFTP上传，仅成功后进入分发通道
            .handle(sftpOutboundAdapter(), 
                    config -> config.advice(deleteFileAdvice())
                                    .successChannel("postUploadDistributeChannel"))
            .get();
}

// 定义上传成功后的分发通道（PublishSubscribeChannel）
@Bean
public PublishSubscribeChannel postUploadDistributeChannel() {
    return new PublishSubscribeChannel();
}

// 归档流程：仅上传成功后触发，按doArchive参数过滤
@Bean
public IntegrationFlow archiveFlow() {
    return IntegrationFlows.from("postUploadDistributeChannel")
            .filter(message -> doArchive) // 替换为你的doArchive参数判断逻辑
            .handle(archiveHandler()) // 你的归档处理器
            .get();
}

// 日志流程：仅上传成功后触发，记录传输成功日志
@Bean
public IntegrationFlow transferLogFlow() {
    return IntegrationFlows.from("postUploadDistributeChannel")
            .handle(message -> {
                // 你的日志记录逻辑，比如记录文件名、传输状态等
                log.info("文件 {} 已成功传输并完成后续处理", message.getPayload());
            })
            .get();
}

// 配置deleteFileAdvice：仅上传成功后执行文件删除
@Bean
public ExpressionEvaluatingRequestHandlerAdvice deleteFileAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setAfterSuccessExpressionString("payload.delete()"); // 替换为你的目标文件删除逻辑
    advice.setPropagateEvaluationFailures(true);
    return advice;
}

方案2：修改PublishSubscribeChannel行为，异常时中断后续订阅者 #

如果你坚持要在同一级分发，需要让PublishSubscribeChannel在第一个订阅者抛出异常时停止后续执行：

1. 禁用异步执行：确保PublishSubscribeChannel使用同步线程执行，让异常能传播到主流程。
2. 不要给出站适配器单独配置errorChannel：让异常直接抛出到主流程，避免被局部通道吞掉。
3. 配置错误处理器，阻止后续分发：

@Bean
public PublishSubscribeChannel distributeChannel() {
    // 不指定taskExecutor，默认使用同步执行
    PublishSubscribeChannel channel = new PublishSubscribeChannel();
    channel.setErrorHandler(interruptingErrorHandler());
    return channel;
}

@Bean
public MessagePublishingErrorHandler interruptingErrorHandler() {
    MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
    errorHandler.setDefaultErrorChannel("globalErrorChannel"); // 全局错误通道处理异常
    // 关键：设置为异常时停止给其他订阅者发送消息
    errorHandler.setSendPartialResultOnError(false);
    return errorHandler;
}

这种方式下，当SFTP上传失败抛出异常时，MessagePublishingErrorHandler会将异常发送到全局错误通道，同时停止将消息分发给后续订阅者，避免误报日志。

额外注意点 #

• 保留轮询器的错误通道配置，用来处理入站阶段的异常（比如拉取远程文件失败、本地目录创建失败等）。
• 在全局错误通道中添加处理器，记录详细的异常栈信息，方便排查SFTP上传失败的具体原因（如权限不足、网络波动、远程路径错误等）。

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