本文介绍如何在 Spring Integration 中对 Sftp.inboundStreamingAdapter 读取的文件流启用异步/并行处理，通过配置线程池和轮询器提升吞吐量，同时避免因共享 InputStream 导致的资源冲突问题。
本文介绍如何在 spring integration 中对 `sftp.inboundstreamingadapter` 读取的文件流启用异步/并行处理，通过配置线程池和轮询器提升吞吐量，同时避免因共享 `inputstream` 导致的资源冲突问题。
在基于 Spring Integration 5.5 + Spring Boot 2.7 的 SFTP 文件处理流程中，若使用 Sftp.inboundStreamingAdapter 直接读取 XML 文件并依次执行解析、持久化与删除操作，默认行为是串行阻塞式处理——每个文件必须等前一个完全结束（包括流关闭）后才开始下一个。这严重限制了 I/O 密集型场景下的吞吐能力。
关键在于：不能在 publishSubscribeChannel 内部强行并行化处理逻辑，尤其当消息体为 InputStream（来自 inboundStreamingAdapter）时。因为多个订阅者可能同时尝试读取或关闭同一远程流，导致 IOException（如 “Stream closed” 或 “Connection reset”），甚至引发 SFTP 会话异常。
✅ 正确解法是将并发控制点前置到消息源头，即让轮询器（SourcePollingChannelAdapter）本身在独立线程中触发每次拉取，并确保每次只拉取一个文件（maxMessagesPerPoll = 1），从而天然隔离各文件的生命周期。
推荐两种等效配置方式：
方式一：通过 endpointConfigurer 配置带线程池的轮询器（推荐）
@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlows.from(
Sftp.inboundStreamingAdapter(sftpTemplate),
e -> e.poller(p -> p
.fixedDelay(1000) // 每秒轮询一次
.maxMessagesPerPoll(1) // 关键！每次仅获取一个文件流
.taskExecutor(taskExecutor()) // 异步执行拉取动作
)
)
.publishSubscribeChannel(spec -> spec
.subscribe(f -> f
.transform(Transformers.fromString()) // 将 InputStream 转为 String
.transform(xmlToDomainObject()) // 解析 XML 为 POJO
.handle((payload, headers) -> repository.save(payload)) // 持久化
)
.subscribe(f -> f
.handle((payload, headers) -> {
// 注意：此处 payload 是原始 Message，需从 headers 获取文件路径
String remotePath = headers.get(FileHeaders.REMOTE_DIRECTORY, String.class)
+ "/" + headers.get(FileHeaders.REMOTE_FILE, String.class);
sftpTemplate.remove(remotePath);
return null;
})
)
)
.get();
}
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(4);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(20);
executor.setThreadNamePrefix("sftp-poller-");
executor.initialize();
return executor;
}
方式二：在 from() 后立即接入线程池通道（更简洁）
@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlows.from(Sftp.inboundStreamingAdapter(sftpTemplate))
.channel(c -> c.executor(taskExecutor())) // 所有后续处理均在该线程池中异步执行
.publishSubscribeChannel(spec -> spec
.subscribe(f -> f.transform(...).handle(...))
.subscribe(f -> f.handle(deleteHandler()))
)
.get();
}
⚠️ 重要注意事项：
-
maxMessagesPerPoll(1)是安全前提：若设为>1，轮询器会在单次调度中批量拉取多个InputStream并同步发出，此时即使有线程池，这些流仍共享同一 SFTP 会话上下文，高并发下易触发连接争用或超时。 - 删除操作必须基于
FileHeaders中的元数据（如REMOTE_FILE,REMOTE_DIRECTORY），切勿依赖InputStream的close()触发删除——流关闭不等于文件已处理完毕，且删除逻辑应与业务处理解耦（如上例中单独订阅）。 - 线程池大小需结合 SFTP 服务器性能、网络延迟及本地处理耗时综合评估；建议初始值
core=4, max=8，再通过监控ThreadPoolTaskExecutor的活跃线程数与队列堆积情况调优。
通过上述任一方式，即可实现“一个文件一个线程”的真正并行处理模型，在保障资源安全的前提下显著提升整体吞吐量。
