Spring中实现SFTP请求-回复功能的技术咨询

需求概述 #

需要实现SFTP请求-回复流程：

• 准备文件并通过SFTP发送到远程服务器
• 同步等待远程服务器返回响应文件（带超时机制）
• 收到响应文件后停止等待、处理数据；超时则标记无响应并恢复流程

现有问题分析 #

1. 异步线程导致流程无法等待：手动注册的SftpInboundAdapter运行在独立线程，主流程process()会直接执行finalLogic()，无法等待响应文件
2. 本地目录触发异常：SftpInboundAdapter同步远程文件到本地后，会监听本地目录的所有变化（包括子目录），不符合仅监听远程特定文件的需求
3. PublishSubscribeChannel并行问题：使用发布订阅通道会同时执行发送和下载逻辑，无法保证先发送再等待回复的顺序

解决方案：基于SftpOutboundGateway的同步请求-回复 #

使用SftpOutboundGateway的get命令主动查询远程文件，配合重试+超时机制实现同步等待，替代异步的SftpInboundAdapter，完美匹配请求-回复场景。

1. 配置SFTP会话工厂（保留现有逻辑） #

@Bean
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    // 配置SFTP连接信息（host、port、username、password等）
    return new CachingSessionFactory<SftpClient.DirEntry>(factory);
}

2. 配置重试通知器（实现超时等待） #

通过RequestHandlerRetryAdvice设置重试间隔和最大重试次数，总超时时间=重试间隔×最大重试次数

@Bean
public RequestHandlerRetryAdvice replyRetryAdvice() {
    RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
    RetryTemplate retryTemplate = new RetryTemplate();
    
    // 设置重试间隔：每10秒检查一次远程文件
    FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
    backOffPolicy.setBackOffPeriod(10000);
    
    // 设置最大重试次数：比如6次，总超时60秒
    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(6);
    
    retryTemplate.setBackOffPolicy(backOffPolicy);
    retryTemplate.setRetryPolicy(retryPolicy);
    
    // 超时后触发的回调：标记无响应状态
    advice.setRetryTemplate(retryTemplate);
    advice.setRecoveryCallback(context -> {
        // 处理超时逻辑，比如返回null或抛出自定义异常
        return null;
    });
    return advice;
}

3. 配置发送+请求回复的整合流程 #

3.1 发送文件的Handler

@Bean
@ServiceActivator(inputChannel = "sftpSendChannel")
public MessageHandler sftpSendHandler(SessionFactory<SftpClient.DirEntry> sftpSessionFactory) {
    SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory);
    handler.setRemoteDirectoryExpressionString("/remoteDir");
    handler.setFileNameGenerator(message -> {
        File file = (File) message.getPayload();
        return file.getName(); // 自定义文件名生成逻辑
    });
    return handler;
}

3.2 等待并下载回复文件的Gateway

@Bean
@ServiceActivator(inputChannel = "sftpReplyChannel")
public MessageHandler sftpReplyHandler(SessionFactory<SftpClient.DirEntry> sftpSessionFactory, RequestHandlerRetryAdvice replyRetryAdvice) {
    // 使用SpEL表达式根据请求文件名生成回复文件名（比如把.txt改成.out）
    SftpOutboundGateway handler = new SftpOutboundGateway(
            sftpSessionFactory, 
            "get", 
            "'/remoteDir/'+payload.getName().replace('.txt','.out')"
    );
    handler.setLocalDirectory(new File("/localDir/reply"));
    handler.setAdviceChain(Collections.singletonList(replyRetryAdvice));
    // 设置文件不存在时抛出异常，触发重试
    handler.setRequiresReply(true);
    return handler;
}

3.3 整合流程的MessagingGateway

用@Gateway的requestChannel和replyChannel实现同步请求-回复，确保先发送文件，再等待回复

@MessagingGateway
public interface SftpRequestReplyGateway {

    @Gateway(requestChannel = "sftpSendChannel", replyChannel = "sftpReplyChannel")
    File requestReply(File requestFile);
}

4. 主流程调用 #

public Status process() {
    // 1. 准备文件
    File requestFile = prepareFile();
    
    // 2. 发送文件并同步等待回复（超时则返回null）
    File replyFile = sftpRequestReplyGateway.requestReply(requestFile);
    
    // 3. 处理结果
    if (replyFile != null) {
        service.processFile(replyFile);
        return Status.SUCCESS;
    } else {
        return Status.NO_RESPONSE;
    }
}

关键说明 #

• 同步等待逻辑：SftpOutboundGateway的get命令会主动查询远程文件，配合重试机制，主线程会阻塞直到文件找到或超时，解决了之前异步线程的问题
• 避免本地目录触发：直接查询远程服务器，无需同步到本地目录再监听，不会触发本地子目录的异常处理
• 顺序保证：通过requestChannel和replyChannel的链式流程，确保先完成文件发送，再开始等待回复，替代了PublishSubscribeChannel的并行执行

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