在Spring Integration Flow中如何正确关闭ExecutorService？

我使用ExecutorService来确保JMS消息写入数据库后再进行确认（目前无法使用XA数据源和分布式事务）。为实现这个需求，我的流程是收到消息后先写入数据库，再通过ExecutorService启动新线程处理后续逻辑。

代码实现如下：

@Bean
public Function<Channels, MessageChannelSpec<?, ?>> jmsTxCommitingChannelSpec() {
    final ExecutorService executorService = Executors.newCachedThreadPool();
    return channels -> channels.executor(executorService);
}
@Bean
public Consumer<HeaderEnricherSpec> errorChannelSpec(MessageChannel genericExceptionChannel) {
    return h -> h.header(MessageHeaders.ERROR_CHANNEL, genericExceptionChannel);
}
@Bean
public IntegrationFlow jmsMessageFlow(
    @Qualifier("jmsConnectionFactory") ConnectionFactory connectionFactory,
    Function<Channels, MessageChannelSpec<?, ?>> jmsTxCommitingChannelSpec)
{
  return IntegrationFlow.from(
            Jms.messageDrivenChannelAdapter(connectionFactory)
                .destination("INCOMING_QUEUE")
                .configureListenerContainer(
                    jmsListenerContainerSpec.andThen(spec -> spec.id("ListenerContainer")))
                .errorChannel(genericExceptionChannel)
                .outputChannel("messageHandlingChannel"))
       // 保存消息到数据库
       .handle(
            (payload, headers) -> databaseService.save(payload),
            spec -> spec.advice(messageRetryAdvice).id("persistClientMessage"))
  // 开启新线程，确保JMS消息能被确认
        .channel(jmsTxCommitingChannelSpec)
        .enrichHeaders(errorChannelSpec)
        .handle(
            (payload, headers) -> messageParser.extractMessageMetadata(payload),
            spec -> spec.id("extractMessageMetadata"))
        .route(incomingMessageRouter)
        .get();
}

我的疑问：在此场景下，我需要关闭ExecutorService吗？如果需要，该怎么操作？

必须关闭ExecutorService，否则应用 shutdown 时会残留非守护线程，导致JVM无法正常退出，甚至引发资源泄漏。下面是具体的优化方案：

1. 将ExecutorService交由Spring管理 #

原来的代码中，ExecutorService是在jmsTxCommitingChannelSpec()内部创建的，Spring无法感知到这个实例，自然也不会帮你关闭它。正确的做法是把它声明为独立的Spring Bean，利用Spring的生命周期管理自动关闭：

// 单独定义ExecutorService Bean，指定destroyMethod为shutdown
@Bean(destroyMethod = "shutdown")
public ExecutorService jmsTxExecutorService() {
    return Executors.newCachedThreadPool();
}

// 注入上面的ExecutorService Bean
@Bean
public Function<Channels, MessageChannelSpec<?, ?>> jmsTxCommitingChannelSpec(ExecutorService jmsTxExecutorService) {
    return channels -> channels.executor(jmsTxExecutorService);
}

Spring会在上下文关闭时自动调用ExecutorService的shutdown()方法，关闭线程池。

2. 优雅关闭（可选但推荐） #

如果需要等待正在执行的任务完成后再关闭，避免任务中途中断，可以自定义销毁逻辑，结合shutdown()和awaitTermination()：

@Bean
public ExecutorService jmsTxExecutorService() {
    return Executors.newCachedThreadPool();
}

@PreDestroy
public void shutdownJmsTxExecutor() {
    ExecutorService executor = jmsTxExecutorService();
    executor.shutdown(); // 拒绝新任务，等待已提交任务完成
    try {
        // 等待60秒让任务完成
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            executor.shutdownNow(); // 强制中断正在执行的任务
            // 再等60秒确认关闭
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("JMS事务线程池无法正常关闭");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt(); // 恢复中断状态
    }
}

@Bean
public Function<Channels, MessageChannelSpec<?, ?>> jmsTxCommitingChannelSpec(ExecutorService jmsTxExecutorService) {
    return channels -> channels.executor(jmsTxExecutorService);
}

3. 为什么不能放任不管？ #

Executors.newCachedThreadPool()创建的是非守护线程，如果不主动关闭，这些线程会一直处于存活状态，即使应用的其他组件都已销毁，JVM也无法正常退出，必须手动kill进程，这在生产环境中是严重的问题。

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