Spring Integration如何并行处理Sftp.inboundStreamingAdapter的消息？

要实现文件的异步并行处理，核心是通过ExecutorChannel将文件处理任务提交到线程池，让多个文件的处理任务可以同时执行。以下是具体修改步骤和代码示例：

1. 定义线程池 #

首先创建一个线程池Bean，参数可根据业务吞吐量调整：

@Bean
public TaskExecutor fileProcessingExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5); // 核心线程数
    executor.setMaxPoolSize(10); // 最大线程数
    executor.setQueueCapacity(20); // 任务队列容量
    executor.setThreadNamePrefix("file-proc-"); // 线程名称前缀，便于日志排查
    executor.initialize();
    return executor;
}

2. 修改IntegrationFlow配置 #

在SFTP源与发布订阅通道之间添加ExecutorChannel，让每个文件的处理流程都在独立线程中执行：

IntegrationFlows
    .from(Sftp.inboundStreamingAdapter(sftpTemplate), 
          e -> e.poller(Pollers.fixedDelay(1000))) // 配置轮询策略，按需调整间隔
    .channel(MessageChannels.executor(fileProcessingExecutor())) // 关键：用线程池实现异步
    .publishSubscribeChannel(spec -> spec
        .subscribe(flow -> flow
            .transform(/* 文件内容转XML */)
            .handle(/* 持久化XML */))
        .subscribe(flow -> flow.handle(/* 删除源文件 */)))
    .get();

关键说明 #

• ExecutorChannel的作用：该通道会将SFTP源拉取到的每个文件消息，提交到指定线程池执行后续流程，实现多个文件的并行处理，直接提升吞吐量。
• Poller配置：Sftp.inboundStreamingAdapter是轮询型源，必须配合poller配置才能正常触发文件拉取，可根据业务需求调整轮询间隔（如fixedDelay(5000)表示每5秒轮询一次）。
• 发布订阅分支的并行性：原有的发布订阅通道会让“XML处理持久化”和“文件删除”两个分支并行执行，如果需要保证先处理完成再删除文件，则需要调整流程结构，将删除操作放在处理分支的末尾，而非独立订阅分支：

  IntegrationFlows
      .from(Sftp.inboundStreamingAdapter(sftpTemplate), e -> e.poller(Pollers.fixedDelay(1000)))
      .channel(MessageChannels.executor(fileProcessingExecutor()))
      .transform(/* 文件内容转XML */)
      .handle(/* 持久化XML */)
      .handle(/* 删除源文件 */)
      .get();
  

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