Spring Integration 文件处理教程

学习目标：掌握 Spring Integration 的文件处理能力，实现高效的文件读写与转换

一、Spring Integration 文件支持概述

Spring Integration 的文件支持模块提供了一套专门用于处理文件操作的集成工具，主要包含以下核心功能：

核心组件

1. FileReadingMessageSource - 文件读取组件
2. FileWritingMessageHandler - 文件写入组件
3. FileTransformer - 文件转换器

实际应用场景

• 监控目录自动处理新文件
• 批量导出数据到文件
• 文件格式转换
• 日志文件实时处理

二、环境配置

1. 添加依赖

build.gradle.kts

dependencies {
    implementation("org.springframework.integration:spring-integration-file:6.5.1")
    implementation("org.springframework.boot:spring-boot-starter-integration")
}

pom.xml

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-file</artifactId>
    <version>6.5.1</version>
</dependency>

2. 基础配置类

@Configuration
@EnableIntegration
class FileIntegrationConfig {

    @Bean
    fun inputChannel(): MessageChannel = DirectChannel()
    
    @Bean
    fun outputChannel(): MessageChannel = DirectChannel()
}

三、文件读取（FileReadingMessageSource）

1. 基础配置示例

@Bean
fun fileReadingMessageSource(): FileReadingMessageSource {
    val source = FileReadingMessageSource().apply {
        setDirectory(File("input"))  // 监控的目录
        setFilter(SimplePatternFileListFilter("*.txt"))  // 文件过滤器
        setScanEachPoll(true)        // 每次轮询扫描新文件
    }
    return source
}

@Bean
fun inboundFileChannelAdapter(source: FileReadingMessageSource): SourcePollingChannelAdapter {
    return IntegrationFlows.from(source)
        .channel("inputChannel")
        .poll(Pollers.fixedDelay(5000))  // 每5秒轮询一次
        .get()
}

TIP

使用 setScanEachPoll(true) 确保每次轮询都能检测到新文件，适用于需要实时处理的场景

2. 文件过滤策略

// 组合过滤器：只接受修改时间在24小时内且大小超过1KB的.txt文件
val filter = CompositeFileListFilter<File>().apply {
    addFilter(LastModifiedFileListFilter(TimeUnit.DAYS.toMillis(1)))
    addFilter(SimplePatternFileListFilter("*.txt"))
    addFilter(SizeFileListFilter(1024))  
}

文件锁定问题

处理文件时使用 FileLocker 防止多个实例同时处理同一文件：

source.setLocker(DefaultFileLocker())

四、文件写入（FileWritingMessageHandler）

1. 基础写入配置

@Bean
fun fileWritingMessageHandler(): FileWritingMessageHandler {
    return FileWritingMessageHandler(File("output")).apply {
        setFileNameGenerator { message ->  // 文件名生成策略
            message.headers["file_name"]?.toString() ?: "default.txt"
        }
        setAppendNewLine(true)  // 每次写入后添加换行符
        setExpectReply(false)   // 不需要回复通道
    }
}

@Bean
fun outboundFlow(): IntegrationFlow {
    return IntegrationFlow.from("outputChannel")
        .handle(fileWritingMessageHandler())
        .get()
}

2. 文件命名策略对比

默认策略

setFileNameGenerator { message ->
    message.headers["file_name"]?.toString() ?: "default.txt"
}

时间戳策略

setFileNameGenerator { message ->
    "output_${System.currentTimeMillis()}.txt"
}

UUID策略

setFileNameGenerator { message ->
    UUID.randomUUID().toString() + ".txt"
}

五、文件转换器

1. 文件转字符串

@Bean
fun fileToStringTransformer(): GenericTransformer<File, String> {
    return GenericTransformer { file -> 
        file.readText(Charsets.UTF_8)  
    }
}

2. 文件转字节数组

@Bean
fun fileToBytesTransformer(): GenericTransformer<File, ByteArray> {
    return GenericTransformer { file ->
        Files.readAllBytes(file.toPath())  
    }
}

3. 在集成流中使用转换器

六、完整文件处理示例

1. 场景描述

• 监控 /input 目录的 .txt 文件
• 转换为大写内容
• 写入 /output 目录并添加时间戳后缀

2. 实现代码

@Configuration
@EnableIntegration
class FileProcessingConfig {

    @Bean
    fun processFlow(): IntegrationFlow {
        return IntegrationFlows
            .from(Files.inboundAdapter(File("input"))
            .filter(FileListFilter { file -> 
                file.extension == "txt" && file.length() > 1024
            })
            .transform(GenericTransformer<File, String> { file ->
                file.readText().uppercase()  // 转换为大写
            })
            .handle(Files.outboundAdapter(File("output"))
            .handle { payload, _ -> 
                File("output/processed_${System.currentTimeMillis()}.txt").apply {
                    writeText(payload as String)
                }
            }
            .get()
    }
}

CAUTION

生产环境应考虑错误处理机制，使用 .handle() 添加错误处理逻辑：

.handle({ payload, _ -> /* 处理逻辑 */ }) { it.advice(retryAdvice()) }

七、最佳实践与常见问题

1. 性能优化技巧

// 使用缓冲读写提高性能
Files.inboundAdapter(directory)
    .usingBuffer(true)  
    .bufferSize(8192)   

// 限制并发处理文件数量
val poller = Pollers.fixedDelay(1000)
    .maxMessagesPerPoll(5)  

2. 常见问题解决

问题1：文件重复处理

// 添加接受一次过滤器
.addFilter(AcceptOnceFileListFilter())

问题2：大文件内存溢出

// 使用分块读取
transform(FileToByteArrayTransformer())
    .splitter(FileSplitter(true))  

问题3：写入权限不足

Files.outboundAdapter(File("/output"))
    .autoCreateDirectory(true)  
    .chmod(644)                 

3. 监控与调试

@Bean
fun loggingHandler(): LoggingHandler {
    return LoggingHandler(LoggingHandler.Level.DEBUG).apply {
        setLoggerName("FILE_PROCESSING")
        setLogExpressionString("'Received: ' + payload")
    }
}

// 在流程中添加日志
...
.channel(ChannelFactories.loggingChannel("beforeTransform"))
.transform(...)
...

总结

Spring Integration 的文件处理模块提供了强大的文件操作能力：

功能	核心组件	使用场景
文件读取	FileReadingMessageSource	监控目录、处理新文件
文件写入	FileWritingMessageHandler	数据导出、结果保存
文件转换	FileTransformer	格式转换、内容处理

::: success 下一步学习

1. 探索与FTP/SFTP模块的集成
2. 学习文件分块处理大文件
3. 了解与Spring Batch的协同工作 :::

关键要点：合理配置轮询策略 + 使用文件锁防止冲突 + 实现健壮的错误处理机制