使用SFTP Outbound Channel Adapter流式上传大文件遇文件名异常

当你用InputStream代替File上传时，Spring Integration的SFTP Outbound Adapter无法从流中获取文件名信息，因此会自动生成随机UUID作为远程文件名，这就是你看到67157c1b-439d-9f3b-f2be-88eff908e39c.msg的原因。同时要实现大文件流式追加上传，需要调整配置传递元数据并适配流式处理逻辑。

1. 传递文件名元数据 #

通过Message<InputStream>包装流数据，在消息头中指定目标文件名，使用FileHeaders.FILENAME作为头键：

@MessagingGateway
public interface MyGateway {
    @Gateway(requestChannel = "toSftpChannel")
    void sendToSftp(Message<InputStream> message);
}

// 调用时构建带文件名的消息
InputStream csvStream = ...; // 你的数据库生成的CSV流
Message<InputStream> message = MessageBuilder.withPayload(csvStream)
        .setHeader(FileHeaders.FILENAME, "bar.txt")
        .build();
gateway.sendToSftp(message);

2. 配置SFTP适配器支持流式追加 #

调整适配器配置，设置FileExistsMode.APPEND启用追加模式，并指定从消息头读取文件名：

@Bean
public IntegrationFlow sftpOutboundFlow(SessionFactory<SftpClient.DirEntry> sftpSessionFactory) {
    return IntegrationFlow.from("toSftpChannel")
            .handle(Sftp.outboundAdapter(sftpSessionFactory, FileExistsMode.APPEND)
                    .useTemporaryFileName(false) // 跳过临时文件，直接写入目标文件
                    .remoteDirectory("upload")
                    .fileNameExpression("headers['file_name']") // 从消息头获取文件名
            ).get();
}

3. 大文件端到端流式处理（适配数据库场景） #

针对从数据库读取大量数据生成CSV并流式上传的场景，建议直接通过SFTP Session获取远程输出流，边生成CSV边写入，完全规避内存溢出风险：

@Autowired
private SessionFactory<SftpClient.DirEntry> sftpSessionFactory;

public void streamCsvToSftp() throws Exception {
    try (Session<SftpClient.DirEntry> session = sftpSessionFactory.getSession()) {
        // 获取远程文件输出流，true表示追加模式
        OutputStream remoteOutputStream = session.write("upload/bar.txt", true);

        // 流式读取数据库并生成CSV
        try (CSVWriter csvWriter = new CSVWriter(new OutputStreamWriter(remoteOutputStream));
             Connection conn = ...; // 获取数据库连接
             Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
             ResultSet rs = stmt.executeQuery("SELECT * FROM large_table")) {

            // 写入CSV表头
            ResultSetMetaData metaData = rs.getMetaData();
            String[] headers = new String[metaData.getColumnCount()];
            for (int i = 0; i < metaData.getColumnCount(); i++) {
                headers[i] = metaData.getColumnName(i + 1);
            }
            csvWriter.writeNext(headers);

            // 逐行写入数据，实时刷新到远程流
            while (rs.next()) {
                String[] row = new String[metaData.getColumnCount()];
                for (int i = 0; i < metaData.getColumnCount(); i++) {
                    row[i] = rs.getString(i + 1);
                }
                csvWriter.writeNext(row);
                csvWriter.flush();
            }
        }
    }
}

关键注意事项 #

• 使用FileExistsMode.APPEND时，远程文件不存在会自动创建，存在则直接追加内容。
• 所有流资源必须通过try-with-resources自动关闭，避免资源泄漏。
• 超大规模数据场景下，可增加数据库查询的分页逻辑，进一步降低内存占用。

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