Spring Integration FTP Outbound：主备FTP主机故障转移配置咨询

要实现主FTP主机A上传失败时自动切换到备机B，你可以通过**Spring Integration的请求处理增强器（Advice）**来捕获A的上传失败，将消息转发到备机B的上传流程，同时统一处理成功后的文件删除，仅当两台主机都失败时才触发错误通知。以下是具体的调整方案：

步骤1：定义两个FTP出站处理器 #

首先为两台FTP主机分别创建出站消息处理器（对应你的主备主机A和B）：

@Bean
public MessageHandler ftpOutboundA() {
    FtpMessageHandler handler = new FtpMessageHandler(ftpSessionFactoryA());
    handler.setRemoteDirectoryExpression(new LiteralExpression("/your/remote/dir/a"));
    handler.setFileNameGenerator(msg -> (String) msg.getHeaders().get(FileHeaders.FILENAME));
    return handler;
}

@Bean
public MessageHandler ftpOutboundB() {
    FtpMessageHandler handler = new FtpMessageHandler(ftpSessionFactoryB());
    handler.setRemoteDirectoryExpression(new LiteralExpression("/your/remote/dir/b"));
    handler.setFileNameGenerator(msg -> (String) msg.getHeaders().get(FileHeaders.FILENAME));
    return handler;
}

注意：这里的ftpSessionFactoryA()和ftpSessionFactoryB()是你已经配置好的FTP会话工厂，对应主备主机的连接信息。

步骤2：修改主IntegrationFlow逻辑 #

调整你的jmsInboundFlow，加入失败切换逻辑：

@Bean
public IntegrationFlow jmsInboundFlow(DefaultMessageListenerContainer myJmsListenerContainer,
                                      MessageHandler ftpOutboundA,
                                      MessageHandler ftpOutboundB) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(myJmsListenerContainer))
            // 尝试上传至主机A，绑定失败处理增强器
            .handle(ftpOutboundA, config -> config.advice(primaryFtpFailureAdvice()))
            // 上传成功后执行本地临时文件删除
            .handle(this::deleteLocalTempFile)
            .get();
}

// 处理主机A上传失败的增强器：将消息转发至备机B的上传流程
@Bean
public Advice primaryFtpFailureAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 构造转发到备机B的消息
    advice.setOnFailureExpressionString(
        "T(org.springframework.messaging.support.MessageBuilder).fromMessage(#root)"
        + ".setHeader('ftp_target', 'backup')"
        + ".build()"
    );
    // 指定失败消息的目标通道（备机B的上传通道）
    advice.setFailureChannel(backupFtpUploadChannel());
    // 不抛出异常，避免中断主流程
    advice.setReturnFailureExpressionResult(false);
    return advice;
}

// 备机B的上传通道
@Bean
public MessageChannel backupFtpUploadChannel() {
    return new DirectChannel();
}

// 备机B的上传流程
@Bean
public IntegrationFlow backupFtpUploadFlow(MessageHandler ftpOutboundB) {
    return IntegrationFlows.from(backupFtpUploadChannel())
            // 上传至备机B，绑定最终失败处理增强器
            .handle(ftpOutboundB, config -> config.advice(finalFtpFailureAdvice()))
            // 上传成功同样执行本地文件删除
            .handle(this::deleteLocalTempFile)
            .get();
}

// 处理备机B上传失败的增强器：触发错误通知
@Bean
public Advice finalFtpFailureAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 构造错误通知消息
    advice.setOnFailureExpressionString(
        "T(org.springframework.messaging.support.MessageBuilder).fromMessage(#root)"
        + ".setHeader('error_msg', '主备FTP上传均失败: ' + #root.cause.message)"
        + ".build()"
    );
    // 指定错误消息的目标通道
    advice.setFailureChannel(errorNotificationChannel());
    // 抛出异常，标记最终失败
    advice.setThrowExceptionOnFailure(true);
    return advice;
}

// 错误通知通道
@Bean
public MessageChannel errorNotificationChannel() {
    return new DirectChannel();
}

// 错误通知处理流程
@Bean
public IntegrationFlow errorNotificationFlow() {
    return IntegrationFlows.from(errorNotificationChannel())
            .handle(msg -> {
                String errorMsg = (String) msg.getHeaders().get("error_msg");
                // 这里实现你的错误信息显示逻辑（比如日志、告警）
                System.err.println("[FTP上传失败] " + errorMsg);
            })
            .get();
}

// 统一的本地临时文件删除方法
private void deleteLocalTempFile(Message<?> message) {
    String localFilePath = (String) message.getHeaders().get(FileHeaders.ORIGINAL_FILE);
    if (localFilePath != null) {
        new File(localFilePath).delete();
    }
}

逻辑说明 #

1. 主流程：从JMS接收消息后，优先上传至主机A；若上传成功，直接执行本地文件删除。
2. 主机A失败处理：通过primaryFtpFailureAdvice捕获失败，将消息转发到备机B的上传通道，不会中断主流程。
3. 备机B上传流程：接收转发的消息，执行上传；若成功同样删除本地文件；若失败则触发错误通知。
4. 最终失败处理：只有当备机B也上传失败时，才会构造错误消息并发送到通知通道，执行错误信息显示。

这种方式既实现了故障自动切换，又保证了成功逻辑的复用，同时仅在最终失败时触发告警，避免了不必要的错误通知。

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