Spring Boot集成SFTP改用ExecutorChannel异步上传失败排查

原本使用DirectChannel实现SFTP上传完全正常，改用ExecutorChannel实现异步并行上传后，文件未成功传输且无异常抛出，相关配置代码如下：

@Bean
public IntegrationFlow sftpOutboundFlow(
        DelegatingSessionFactory lDelegatingSessionFactory)
{
    getRemotePath(lDelegatingSessionFactory);
    return IntegrationFlow.from("outboundSftpChannel")
            .enrichHeaders(lHeader -> lHeader.header("remoteDirectory",
                    lRemoteDirectoryWithHostName))
            .enrichHeaders(h -> h.header(MessageHeaders.ERROR_CHANNEL, "errorChannel"))   //Here i have added errorChannel 
            .handle(Sftp.outboundAdapter(lDelegatingSessionFactory,
                    FileExistsMode.REPLACE)
                    .remoteDirectoryExpression
                            ("headers['remoteDirectory']")).get();
}

@Bean
public MessageChannel outboundSftpChannel()
{
  //  return new DirectChannel();   // it was working fine ,below is the new implementation 
       ExecutorChannel executorChannel = new ExecutorChannel(taskExecutor());
   // executorChannel.addInterceptor(new ThreadLoggingInterceptor());
    return executorChannel;
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);  // Minimum number of threads in the pool
    executor.setMaxPoolSize(10);  // Maximum number of threads in the pool
    executor.setQueueCapacity(25);  // Queue capacity for pending tasks
    executor.setThreadNamePrefix("AsyncExecutor-");  // Prefix for thread names
    executor.setWaitForTasksToCompleteOnShutdown(true);  // Ensures tasks complete on shutdown
    executor.setAwaitTerminationSeconds(60);  // Timeout for waiting for tasks to complete
    executor.initialize();  // Initializes the thread pool
    return executor;
}

1. SessionFactory线程安全性问题 #

• 原因：DelegatingSessionFactory本身线程安全，但内部目标SessionFactory（如默认DefaultSftpSessionFactory）是非线程安全的，多线程并行访问会导致Session混乱，上传失败但无明显异常。
• 解决办法：使用CachingSessionFactory包装目标SessionFactory，保证线程间Session隔离：

@Bean
public CachingSessionFactory cachingSessionFactory(DefaultSftpSessionFactory targetFactory) {
    return new CachingSessionFactory(targetFactory, 10); // 缓存大小匹配线程池最大数
}

将IntegrationFlow中的DelegatingSessionFactory替换为CachingSessionFactory。

2. 错误通道未处理异常 #

• 原因：配置了ERROR_CHANNEL头，但errorChannel无对应处理器，异常被静默丢弃，无法感知上传失败。
• 解决办法：定义errorChannel的处理逻辑，记录异常详情：

@Bean
public IntegrationFlow errorHandlingFlow() {
    return IntegrationFlow.from("errorChannel")
            .handle(message -> {
                MessagingException exception = (MessagingException) message.getPayload();
                System.err.println("SFTP上传失败: " + exception.getMessage());
                exception.printStackTrace();
            })
            .get();
}

3. 线程池初始化冲突 #

• 原因：手动调用executor.initialize()与Spring容器生命周期管理冲突，可能导致线程池未正确初始化；或应用关闭时任务未完成被强制终止。
• 解决办法：移除executor.initialize()调用，Spring会自动处理线程池初始化；确认线程池的 shutdown 配置生效，避免任务被中断。

4. 远程目录变量线程不安全 #

• 原因：lRemoteDirectoryWithHostName如果是共享变量，多线程环境下可能被并发修改，导致上传到错误目录或目录不存在，适配器可能静默处理目录创建失败。
• 解决办法：确保lRemoteDirectoryWithHostName线程安全，或改用表达式直接计算远程目录：

.enrichHeaders(lHeader -> lHeader.headerExpression("remoteDirectory", 
    "@remotePathProvider.getRemotePath(#root.sessionFactory)"))

其中remotePathProvider是线程安全的Bean，负责根据SessionFactory获取对应远程目录。

5. 线程池拒绝策略导致消息丢失 #

• 原因：线程池队列满时，默认拒绝策略会丢弃任务，无异常抛出。
• 解决办法：配置合理的拒绝策略，避免消息丢失：

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

同时启用Spring Integration消息跟踪，监控消息流转：

spring.integration.message-tracking.enabled=true

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