Spring中如何并行执行两个SQS消费者Bean的销毁操作

我有一个Spring应用，启动了两个各自持有ExecutorService的SQS消费者，每个消费者会启动N个无限轮询循环。由于是SQS轮询器，需要在进程管理器发送SIGTERM/SIGINT信号时实现优雅关闭，但目前两个消费者是串行销毁的，导致其中一个在处理时另一个仍在轮询。经排查发现两个消费者最终都会终止，但会按顺序执行，若遇到超时较长的消费者，需等待其完成才会触发另一个的销毁。请问如何实现并行销毁？

相关代码 #

SQSManagerA 实现 #

class SQSManagerA(
    threadPoolSize: Int,
    handler: EC2SQSEventHandler,
    sqsConsumerEnable: Boolean,
    backOffRetry: RetryStrategy<Void>
) {
    private val executor: ExecutorService
    private var pollers: ArrayList<PollingLoop>
    private val waitForTermination = CountDownLatch(1)
    private val log = LoggerFactory.getLogger(SQSManagerA::class.java)

    init {
        executor = Executors.newFixedThreadPool(threadPoolSize)
        pollers = ArrayList()
        if (sqsConsumerEnable) {
            for (i: Int in IntStream.range(0, threadPoolSize)) {
                val poller = PollingLoop(handler, backOffRetry)
                pollers.add(poller)
                executor.execute(poller)
            }
        }
    }

    fun destroy() {
        log.info("Received a shutdown signal. Signal stop on all threads and wait for 60 seconds")
        for (poller in pollers) {
            poller.stop()
        }
        waitForTermination.await(delayElbTargetPoller, TimeUnit.SECONDS)
        log.info("Final shutdown on all threads.")
        executor.shutdownNow()
    }
}

PollingLoopA 实现 #

poller.stop()仅通过修改布尔标志让SQS工作线程退出无限循环，对应的PollingLoop代码如下：

class PollingLoopA(
    private val queue: NotificationQueueHandler<>,
    private val handler: EventHandler<>,
) : Runnable {
    @Volatile
    private var terminate = false

    override fun run() {
        log.info(queue.sourceQueueName + " polling loop is running")
        var failedAttempts = 0
        while (!terminate) {
            //work
        }
        log.info("${currentThread().name} has stopped polling.")
    }

    fun stop() {
        log.info("${currentThread().name} has called to stop polling.")
        this.terminate = true
    }
}

销毁日志（串行执行现象） #

[2022-07-23T21:52:57,445] [INFO] (QueueA Polling Thread 28) {} PollingLoopA: No messageA notifications in the queue.
[2022-07-23T21:52:57,475] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.
[2022-07-23T21:53:17,621] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.
[2022-07-23T21:53:17,629] [INFO] (QueueA Polling Thread 28) {} PollingLoopA: No messageA notifications in the queue.
[2022-07-23T21:53:18,413] [INFO] (SIGINT handler) {} spring.Launcher: Initiating shutdown
[2022-07-23T21:53:18,414] [INFO] (SIGINT handler) {} SQSManagerA: Received a shutdown signal. Signal stop on all pollers and wait for 300 seconds
[2022-07-23T21:53:18,415] [INFO] (SIGINT handler) {} PollingLoopA: SIGINT handler has called to stop polling.
[2022-07-23T21:53:37,814] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.
[2022-07-23T21:53:37,828] [INFO] (QueueA Polling Thread 28) {} PollingLoopA: No messageA notifications in the queue.
[2022-07-23T21:53:37,909] [INFO] (QueueA Polling Thread 28) {} PollingLoopA: QueueA Polling Thread 28 has stopped polling.
[2022-07-23T21:53:57,970] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.
[2022-07-23T21:54:18,125] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.
[2022-07-23T21:54:38,273] [INFO] (QueueB Polling Thread 25) {} PollingLoopB: No messageB notifications in the queue.

解决方案：并行触发销毁并等待所有消费者完成 #

1. 修正单个SQSManager的销毁逻辑 #

当前destroy()方法存在两个核心问题：一是waitForTermination计数逻辑无效（初始值为1但从未触发countDown）；二是Spring默认串行调用多个Manager的销毁方法。先修复单个Manager的内部等待逻辑：

调整SQSManager的CountDownLatch初始化

将计数设置为轮询器数量，确保所有轮询器退出后才完成等待：

class SQSManagerA(
    threadPoolSize: Int,
    handler: EC2SQSEventHandler,
    sqsConsumerEnable: Boolean,
    backOffRetry: RetryStrategy<Void>
) {
    private val executor: ExecutorService
    private var pollers: ArrayList<PollingLoop>
    private val waitForTermination: CountDownLatch
    private val log = LoggerFactory.getLogger(SQSManagerA::class.java)

    init {
        executor = Executors.newFixedThreadPool(threadPoolSize)
        pollers = ArrayList()
        if (sqsConsumerEnable) {
            waitForTermination = CountDownLatch(threadPoolSize)
            for (i: Int in IntStream.range(0, threadPoolSize)) {
                val poller = PollingLoop(handler, backOffRetry, waitForTermination)
                pollers.add(poller)
                executor.execute(poller)
            }
        } else {
            waitForTermination = CountDownLatch(1)
            waitForTermination.countDown()
        }
    }

    fun destroy() {
        log.info("Received a shutdown signal. Signal stop on all threads and wait for 60 seconds")
        // 并行通知所有轮询器停止
        pollers.forEach { it.stop() }
        // 等待所有轮询器退出
        val terminated = waitForTermination.await(60, TimeUnit.SECONDS)
        if (!terminated) {
            log.warn("Not all pollers stopped within timeout, forcing shutdown")
        }
        executor.shutdownNow()
        log.info("Final shutdown on all threads.")
    }
}

修改PollingLoop，退出时触发Latch计数

class PollingLoopA(
    private val queue: NotificationQueueHandler<>,
    private val handler: EventHandler<>,
    private val terminationLatch: CountDownLatch
) : Runnable {
    @Volatile
    private var terminate = false

    override fun run() {
        log.info(queue.sourceQueueName + " polling loop is running")
        var failedAttempts = 0
        try {
            while (!terminate) {
                //work
            }
        } finally {
            log.info("${currentThread().name} has stopped polling.")
            terminationLatch.countDown()
        }
    }

    fun stop() {
        log.info("${currentThread().name} has called to stop polling.")
        this.terminate = true
    }
}

2. 实现多SQSManager的并行销毁 #

Spring默认串行调用Bean销毁方法，需自定义管理器实现并行触发：

@Component
class ShutdownManager(
    private val sqsManagers: List<SQSManagerA>,
    @Value("\${shutdown.timeout:60}") private val shutdownTimeout: Long
) : DisposableBean {

    private val log = LoggerFactory.getLogger(ShutdownManager::class.java)
    private val executor = Executors.newFixedThreadPool(sqsManagers.size)

    override fun destroy() {
        log.info("Initiating parallel shutdown of all SQS managers")
        // 并行提交所有销毁任务
        val futures = sqsManagers.map { manager ->
            executor.submit { manager.destroy() }
        }
        // 等待所有销毁任务完成
        futures.forEach { future ->
            try {
                future.get(shutdownTimeout, TimeUnit.SECONDS)
            } catch (e: Exception) {
                log.error("SQS manager shutdown timed out or failed", e)
            }
        }
        executor.shutdown()
        log.info("All SQS managers have been shut down")
    }
}

3. 优化轮询响应速度 #

SQS长轮询（默认20秒）会导致terminate标志生效延迟，建议在轮询逻辑中支持线程中断：

override fun run() {
    log.info(queue.sourceQueueName + " polling loop is running")
    var failedAttempts = 0
    try {
        while (!terminate && !Thread.currentThread().isInterrupted) {
            try {
                // 调用SQS接收消息，响应中断
                val messages = sqsClient.receiveMessage(receiveMessageRequest)
                // 处理消息
            } catch (e: InterruptedException) {
                terminate = true
                Thread.currentThread().interrupt() // 保留中断标志
            } catch (e: Exception) {
                // 处理业务异常
            }
        }
    } finally {
        log.info("${currentThread().name} has stopped polling.")
        terminationLatch.countDown()
    }
}

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