Spring Integration：SFTP出站网关GET(STREAM)并发访问与队列响应获取

1. 文件名Header验证方案 #

带STREAM选项的GET命令返回的消息确实包含文件名相关Header，可用于验证流与请求文件的对应关系：
Spring Integration的SFTP出站网关在处理GET（STREAM模式）请求时，会自动为返回消息添加以下关键Header：

• FileHeaders.FILENAME（字符串常量为file_name）：存储远程文件的文件名
• FileHeaders.REMOTE_FILE（字符串常量为remote_file）：存储远程文件的完整路径

你可以在获取流时先验证Header，确保拿到的是自身请求的流，示例代码如下：

private InputStream getInputStream(String expectedFileName) {
    Message<?> msg = stream.receive(ftpTimeout);
    if (msg == null) {
        return null;
    }
    // 验证远程文件路径是否匹配目标文件名
    String remoteFilePath = (String) msg.getHeaders().get(FileHeaders.REMOTE_FILE);
    if (remoteFilePath != null && remoteFilePath.endsWith(expectedFileName)) {
        return (InputStream) msg.getPayload();
    } else {
        // 不匹配则将消息放回队列，避免丢失其他请求的响应
        stream.send(msg);
        return null;
    }
}

调用时传入请求的filename即可完成校验，避免拿错其他请求的流。

2. 自定义响应通道方案 #

Spring Integration原生支持调用端指定专属响应通道，这是解决并发竞争最可靠的方案，完全不需要共享全局队列：

实现步骤： #

1. 调用时创建专属临时队列通道，通过MessageBuilder将其设置为请求消息的ReplyChannelHeader：

public InputStream openFileStream(final int retryCount, final String filename, final String directory) throws Exception {
    InputStream is = null;
    // 为当前请求创建专属内存队列通道
    QueueChannel exclusiveReplyChannel = MessageChannels.queue().get();
    String remoteFilePath = directory + "/" + filename;
    
    for (int i = 1; i <= retryCount; ++i) {
        Message<String> requestMsg = MessageBuilder.withPayload(remoteFilePath)
                .setReplyChannel(exclusiveReplyChannel) // 指定专属响应通道
                .build();
        if (sftpGetInputChannel.send(requestMsg, ftpTimeout)) {
            Message<?> replyMsg = exclusiveReplyChannel.receive(ftpTimeout);
            if (replyMsg != null) {
                is = (InputStream) replyMsg.getPayload();
                break;
            } else {
                logger.info("Failed to obtain input stream so attempting retry " + i + " of " + retryCount);
                Thread.sleep(ftpTimeout);
            }
        }
    }
    return is;
}

2. 移除原配置中的全局remoteFileOutputChannel，SFTP出站网关会自动将响应发送到请求消息指定的专属通道中。

优势： #

• 每个请求拥有独立的响应通道，从根源上避免了共享队列的竞争问题
• 无需额外同步或校验逻辑，完全符合Spring Integration的请求-响应设计模式
• 临时内存队列由Spring自动管理生命周期，无需手动销毁

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