Spring Integration FTP 出站网关详解

本教程将详细介绍 Spring Integration 中的 FTP 出站网关，帮助初学者掌握通过 Spring 应用与 FTP 服务器交互的核心技术。所有示例均使用 Kotlin 编写，采用现代 Spring 最佳实践。

一、FTP 出站网关概述

1.1 基本概念与作用

FTP 出站网关是 Spring Integration 提供的核心组件，用于在 Spring 应用中与远程 FTP/FTPS 服务器进行交互。它提供了一套标准化命令，简化了文件传输操作，使开发者无需直接处理底层的 FTP 协议细节。

为什么选择出站网关？

✅ 简化开发：封装复杂 FTP 操作
✅ 统一接口：提供一致的编程模型
✅ 集成便捷：无缝融入 Spring Integration 流

1.2 支持的核心命令

命令	描述	使用场景
`ls`	列出远程文件	目录内容查看
`nlst`	列出文件名	简单文件名获取
`get`	下载单个文件	文件获取
`mget`	批量下载文件	批量文件同步
`put`	上传单个文件	文件上传
`mput`	批量上传文件	批量文件上传
`rm`	删除文件	文件清理
`mv`	移动/重命名文件	文件管理

二、核心命令详解

2.1 文件列表命令

2.1.1 `ls` 命令

kotlin

@Bean
fun ftpOutboundGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("listChannel")
        .handle(Ftp.outboundGateway(sf, Command.LS, "'remote_dir/'")
        .get()
}

选项	描述	示例
`-1`	仅返回文件名	`ls -1`
`-a`	包含隐藏文件	`ls -a`
`-dirs`	包含目录	`ls -dirs`
`-R`	递归列出	`ls -R`

TIP

使用 -R 递归选项时，fileName 包含子目录路径，形成相对路径结构

2.1.2 `nlst` 命令

kotlin

@Bean
fun nlstGateway(sf: SessionFactory<FTPFile>): MessageHandler {
    return Ftp.outboundGateway(sf, Command.NLST, "headers['targetDir']")
        .commandOptions(" -f") // 不排序
        .outputChannelName("fileListChannel")
}

CAUTION

nlst 命令仅返回文件名，不包含文件详细信息（如类型、大小等），适用于简单文件名获取场景

2.2 文件下载命令

2.2.1 `get` 命令

kotlin

@Bean
fun downloadGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("downloadChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.GET, "payload")
                .localDirectoryExpression("'local_dir/' + #remoteDirectory")
                .localFilenameGeneratorExpression("#remoteFileName.toUpperCase()")
                .options(Option.STREAM)
        )
        .channel("fileProcessingChannel")
        .get()
}

选项	描述
`-P`	保留远程文件时间戳
`-stream`	以流形式获取文件
`-D`	下载后删除远程文件

2.2.2 `mget` 命令

kotlin

@Bean
fun batchDownloadGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("batchDownloadChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.MGET, "payload + '/*.txt'")
                .localDirectoryExpression("'local_dir/' + #remoteDirectory")
                .options(Option.RECURSIVE, Option.PRESERVE_TIMESTAMP)
                .filter(RegexFileListFilter("^.*\\.txt$").apply {
                    alwaysAcceptDirectories = true
                })
        )
        .channel("batchFilesChannel")
        .get()
}

IMPORTANT

使用递归选项 -R 时，模式会被忽略，默认检索整个目录树。务必设置合适的文件过滤器！

2.3 文件上传命令

2.3.1 `put` 命令

kotlin

@Bean
fun uploadGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("uploadChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.PUT, "headers['remoteDir']")
                .remoteFileNameGenerator(DefaultFileNameGenerator().apply {
                    setExpression("headers['customFilename'] ?: payload.name")
                })
                .chmod(600) // 设置文件权限
        )
        .get()
}

文件权限说明

chmod 参数使用标准的 Unix 八进制格式：

600：所有者读写权限
644：所有者读写，其他用户读
700：所有者所有权限

2.3.2 `mput` 命令

kotlin

@Bean
fun batchUploadGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("batchUploadChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.MPUT, "payload")
                .remoteDirectoryExpression("headers['targetDir']")
                .mputFilter(CompositeFileListFilter<File>().apply {
                    addFilter(RegexFileListFilter("^.*\\.(txt|csv)$"))
                })
                .chmod(644)
        )
        .get()
}

2.4 文件管理命令

2.4.1 `rm` 命令

kotlin

@Bean
fun deleteGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("deleteChannel")
        .handle(Ftp.outboundGateway(sf, Command.RM, "payload"))
        .get()
}

2.4.2 `mv` 命令

kotlin

@Bean
fun moveGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("moveChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.MV, "payload")
                .renameExpression("headers['newPath']")
        )
        .get()
}

三、高级配置与最佳实践

3.1 Kotlin DSL 配置

kotlin

@Configuration
class FtpConfig {

    @Bean
    fun ftpSessionFactory(): SessionFactory<FTPFile> {
        return DefaultFtpSessionFactory().apply {
            host = "ftp.example.com"
            port = 21
            username = "user"
            password = "pass"
            isTestSession = true
        }.let { CachingSessionFactory(it) }
    }

    @Bean
    fun ftpFlow(gateway: FtpOutboundGatewaySpec): IntegrationFlow {
        return IntegrationFlow.from("ftpInChannel")
            .handle(gateway)
            .channel(MessageChannels.queue("resultChannel"))
            .get()
    }

    @Bean
    fun ftpOutboundGateway(): FtpOutboundGatewaySpec {
        return Ftp.outboundGateway(ftpSessionFactory(), Command.MGET, "payload")
            .options(Option.RECURSIVE)
            .regexFileNameFilter("(subDir|.*1.txt)")
            .localDirectoryExpression("'local/' + #remoteDirectory")
            .localFilenameExpression("#remoteFileName.replace('ftp', 'local')")
    }
}

3.2 部分成功处理

当批量操作中出现部分失败时，Spring 会抛出 PartialSuccessException：

kotlin

@ServiceActivator(inputChannel = "errorChannel")
fun handlePartialSuccess(exception: MessagingException) {
    if (exception is PartialSuccessException) {
        val successful = exception.partialResults // 成功文件列表
        val failed = exception.derivedInput - successful // 失败文件列表
        
        logger.info("部分成功操作: ${successful.size} 成功, ${failed.size} 失败")
        // 实现重试逻辑...
    }
}

3.3 文件过滤策略

kotlin

@Bean
fun customFileFilter(): FileListFilter<FTPFile> {
    return CompositeFileListFilter<FTPFile>().apply {
        addFilter(RegexFileListFilter("^.*\\.(pdf|docx)$")) // 按扩展名过滤
        addFilter(LastModifiedFileListFilter().apply {
            age = 60 // 仅最近60秒修改的文件
        })
    }
}

@Bean
fun filteredGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("filteredChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.MGET, "payload")
                .filter(customFileFilter())
        )
        .get()
}

四、常见问题解决方案

4.1 连接问题排查

kotlin

@Bean
fun debugSessionFactory(): SessionFactory<FTPFile> {
    return object : DefaultFtpSessionFactory() {
        override fun postProcessClientAfterConnect(client: FTPClient) {
            client.addProtocolCommandListener(object : PrintCommandListener(
                PrintWriter(System.out), true) 
            )
        }
    }
}

4.2 大文件处理优化

kotlin

@Bean
fun streamDownloadGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("largeFileChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.GET, "payload")
                .options(Option.STREAM) 
        )
        .handle { message, _ ->
            message.headers["closeableResource"]?.let {
                if (it is Closeable) it.close() // 必须手动关闭流
            }
            message
        }
        .split(FileSplitter(true)) // 流式处理大文件
        .channel("lineProcessingChannel")
        .get()
}

WARNING

使用流模式时，必须确保在文件处理完成后手动关闭会话资源，否则会导致连接泄漏

4.3 递归操作优化

kotlin

@Bean
fun recursiveFilter(): FileListFilter<FTPFile> {
    return FtpRegexPatternFileListFilter("^.*\\.txt$").apply {
        alwaysAcceptDirectories = true // 允许递归目录
    }
}

@Bean
fun recursiveGateway(sf: SessionFactory<FTPFile>): IntegrationFlow {
    return IntegrationFlow.from("recursiveChannel")
        .handle(
            Ftp.outboundGateway(sf, Command.MGET, "payload")
                .options(Option.RECURSIVE)
                .filter(recursiveFilter())
        )
        .get()
}

五、最佳实践总结

连接管理
- 使用 CachingSessionFactory 复用连接
- 设置合理的连接超时时间
- 实现连接健康检查

错误处理

kotlin

@Bean
fun errorHandlingFlow(): IntegrationFlow {
    return IntegrationFlow.from("errorChannel")
        .handle { ex: MessagingException ->
            when (val cause = ex.cause) {
                is FTPConnectionClosedException -> 
                    logger.error("连接意外关闭", cause)
                is FTPDataTransferException -> 
                    logger.error("文件传输错误", cause)
                else -> logger.error("未知错误", ex)
            }
        }
        .get()
}

性能优化
- 批量操作使用 mget/mput 替代多次 get/put
- 大文件使用流式传输
- 启用 FTP 被动模式减少防火墙问题

安全增强

kotlin

@Bean
fun secureSessionFactory(): SessionFactory<FTPSFile> {
    return DefaultFtpsSessionFactory().apply {
        host = "secure.example.com"
        port = 990
        protocol = "TLSv1.3"
        useClientMode = true
        implicit = true
    }
}

迁移指南

从 XML 配置迁移到 Kotlin DSL：

将 <int-ftp:outbound-gateway> 替换为 Ftp.outboundGateway()
属性转换为对应的方法调用（如 command-options → options()）
过滤器配置转换为 Kotlin 代码

通过本教程，您应该掌握了 Spring Integration FTP 出站网关的核心用法。在实际应用中，建议结合具体业务需求选择合适的命令和配置策略，并遵循最佳实践确保系统的稳定性和性能。

Spring Integration FTP 出站网关详解 ​

一、FTP 出站网关概述 ​

1.1 基本概念与作用 ​

1.2 支持的核心命令 ​

二、核心命令详解 ​

2.1 文件列表命令 ​

2.1.1 ls 命令 ​

2.1.2 nlst 命令 ​

2.2 文件下载命令 ​

2.2.1 get 命令 ​

2.2.2 mget 命令 ​

2.3 文件上传命令 ​

2.3.1 put 命令 ​

2.3.2 mput 命令 ​

2.4 文件管理命令 ​

2.4.1 rm 命令 ​

2.4.2 mv 命令 ​

三、高级配置与最佳实践 ​

3.1 Kotlin DSL 配置 ​

3.2 部分成功处理 ​

3.3 文件过滤策略 ​

四、常见问题解决方案 ​

4.1 连接问题排查 ​

4.2 大文件处理优化 ​

4.3 递归操作优化 ​

五、最佳实践总结 ​

Spring Integration FTP 出站网关详解

一、FTP 出站网关概述

1.1 基本概念与作用

1.2 支持的核心命令

二、核心命令详解

2.1 文件列表命令

2.1.1 `ls` 命令

2.1.2 `nlst` 命令

2.2 文件下载命令

2.2.1 `get` 命令

2.2.2 `mget` 命令

2.3 文件上传命令

2.3.1 `put` 命令

2.3.2 `mput` 命令

2.4 文件管理命令

2.4.1 `rm` 命令

2.4.2 `mv` 命令

三、高级配置与最佳实践

3.1 Kotlin DSL 配置

3.2 部分成功处理

3.3 文件过滤策略

四、常见问题解决方案

4.1 连接问题排查

4.2 大文件处理优化

4.3 递归操作优化

五、最佳实践总结