Spring Integration结合TaskExecutor优雅停机问题及调度线程疑问

一、修复Poller自定义TaskExecutor的优雅停机问题 #

当为Poller指定自定义ThreadPoolTaskExecutor时，Spring Integration不会自动将其纳入Spring Boot的优雅停机生命周期管理，导致Tomcat关闭后任务仍在执行。可通过以下方式解决：

1. 用IntegrationManagedTaskExecutor包装自定义Executor
   将你的ThreadPoolTaskExecutor包装为Spring Integration提供的IntegrationManagedTaskExecutor，它会自动参与应用生命周期回调，停机时等待任务完成：

   @Bean
   public TaskExecutor fileProcessingExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
       executor.setCorePoolSize(5);
       executor.setMaxPoolSize(10);
       executor.setWaitForTasksToCompleteOnShutdown(true);
       executor.setAwaitTerminationSeconds(120);
       executor.setThreadNamePrefix("file-processor-");
       executor.initialize();
       return new IntegrationManagedTaskExecutor(executor);
   }
   

   然后在Poller配置中使用该包装后的Executor：

   @Bean
   public PollerSpec filePoller() {
       return Pollers.fixedDelay(Duration.ofSeconds(10))
               .taskExecutor(fileProcessingExecutor())
               .maxMessagesPerPoll(5);
   }
   

2. 手动将Executor纳入生命周期管理
   若不想使用IntegrationManagedTaskExecutor，可通过@PreDestroy方法手动触发Executor终止：

   @Bean
   public ThreadPoolTaskExecutor fileProcessingExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
       executor.setCorePoolSize(5);
       executor.setWaitForTasksToCompleteOnShutdown(true);
       executor.setAwaitTerminationSeconds(120);
       executor.setThreadNamePrefix("file-processor-");
       executor.initialize();
       return executor;
   }
   
   @PreDestroy
   public void shutdownExecutor() {
       ThreadPoolTaskExecutor executor = fileProcessingExecutor();
       executor.shutdown();
       try {
           if (!executor.awaitTermination(120, TimeUnit.SECONDS)) {
               executor.shutdownNow();
           }
       } catch (InterruptedException e) {
           executor.shutdownNow();
       }
   }
   

   同时确保Spring Boot优雅停机配置已开启：

   spring.lifecycle.timeout-per-shutdown-phase=120s
   spring.shutdown.graceful.enabled=true
   

二、修复Spring Task Scheduling的多线程与停机等待问题 #

改用Spring Task Scheduling后出现单线程运行、停机等待REST调用的问题，可按以下步骤修复：

1. 正确配置TaskScheduler并关联到Poller
   自定义TaskScheduler并确保Poller使用该调度器，替代默认单线程实例：

   @Bean
   public ThreadPoolTaskScheduler taskScheduler() {
       ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
       scheduler.setPoolSize(2);
       scheduler.setWaitForTasksToCompleteOnShutdown(true);
       scheduler.setAwaitTerminationSeconds(120);
       scheduler.setThreadNamePrefix("scheduler-");
       return scheduler;
   }
   
   @Bean
   public PollerSpec filePoller() {
       return Pollers.fixedDelay(Duration.ofSeconds(10))
               .taskScheduler(taskScheduler())
               .maxMessagesPerPoll(2);
   }
   

   注意：maxMessagesPerPoll的值不要超过线程池大小，避免任务堆积。

2. 处理REST调用的优雅关闭
   若文件处理逻辑包含REST调用，需确保客户端支持优雅关闭，避免停机时线程阻塞：

   • 对于RestTemplate，在@PreDestroy中关闭底层请求工厂：

     @Autowired
     private RestTemplate restTemplate;
     
     @PreDestroy
     public void shutdownRestTemplate() {
         ClientHttpRequestFactory factory = restTemplate.getRequestFactory();
         if (factory instanceof Closeable) {
             try {
                 ((Closeable) factory).close();
             } catch (IOException e) {
                 // 异常处理
             }
         }
     }
     

   • 对于WebClient，配置Reactor HttpClient的优雅关闭：

     @Bean
     public WebClient webClient() {
         HttpClient httpClient = HttpClient.create()
                 .doOnDispose(() -> {
                     try {
                         httpClient.disposeLater().block(Duration.ofSeconds(30));
                     } catch (Exception e) {
                         // 异常处理
                     }
                 });
         return WebClient.builder()
                 .clientConnector(new ReactorClientHttpConnector(httpClient))
                 .build();
     }
     

3. 调整停机顺序
   Spring Boot默认停机顺序可能导致Tomcat先关闭，而Spring Integration任务仍在运行。可通过@Order调整Bean销毁顺序，优先终止Integration任务：

   @Bean
   @Order(Ordered.HIGHEST_PRECEDENCE)
   public ApplicationListener<ContextClosedEvent> integrationShutdownListener() {
       return event -> {
           IntegrationContextUtils.getIntegrationContext(event.getApplicationContext())
                   .stop(Runnable::run);
       };
   }
   

关键注意事项 #

• 所有自定义Executor/Scheduler需设置waitForTasksToCompleteOnShutdown=true和awaitTerminationSeconds=120，与优雅超时时间一致。
• 避免在文件处理逻辑中添加无法中断的阻塞操作，否则shutdownNow()无法终止线程，导致超时后强制关闭。
• 建议升级Spring Boot到最新稳定版本（如2.7.x或3.x），旧版本可能存在Integration优雅停机的Bug。

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