From 55d1289a355b35a7f59526e0c47649484b41e6d2 Mon Sep 17 00:00:00 2001 From: Mat Jaggard Date: Wed, 16 Mar 2022 12:37:26 +0000 Subject: [PATCH] GH-1436: Async Stop Containers Resolves https://github.com/spring-projects/spring-amqp/issues/1436 Allow shutdown to be started but waiting to be completed asynchronously Use Task Executor from parent Update abstract parent to allow running to be set to false --- .../AbstractMessageListenerContainer.java | 30 ++--- .../SimpleMessageListenerContainer.java | 105 +++++++++++------- 2 files changed, 76 insertions(+), 59 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 8f0dfe69..17b2beb1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -109,6 +109,7 @@ import io.micrometer.core.instrument.Timer.Sample; * @author Arnaud Cogoluègnes * @author Artem Bilan * @author Mohammad Hewedy + * @author Mat Jaggard */ public abstract class AbstractMessageListenerContainer extends RabbitAccessor implements MessageListenerContainer, ApplicationContextAware, BeanNameAware, DisposableBean, @@ -1331,10 +1332,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor throw convertRabbitAccessException(ex); } finally { - synchronized (this.lifecycleMonitor) { - this.running = false; - this.lifecycleMonitor.notifyAll(); - } + setNotRunning(); + } + } + + protected void setNotRunning() { + synchronized (this.lifecycleMonitor) { + this.running = false; + this.lifecycleMonitor.notifyAll(); } } @@ -1420,20 +1425,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor throw convertRabbitAccessException(ex); } finally { - synchronized (this.lifecycleMonitor) { - this.running = false; - this.lifecycleMonitor.notifyAll(); - } - } - } - - @Override - public void stop(Runnable callback) { - try { - stop(); - } - finally { - callback.run(); + setNotRunning(); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index f0cd3ee9..ed9cfc51 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -77,6 +77,7 @@ import com.rabbitmq.client.ShutdownSignalException; * @author Gary Russell * @author Artem Bilan * @author Alex Panchenko + * @author Mat Jaggard * * @since 1.0 */ @@ -605,59 +606,83 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta @Override protected void doShutdown() { + shutdownAndWaitOrCallback(null); + } + + @Override + public void stop(Runnable callback) { + shutdownAndWaitOrCallback(() -> { + setNotRunning(); + callback.run(); + }); + } + + private void shutdownAndWaitOrCallback(@Nullable Runnable callback) { Thread thread = this.containerStoppingForAbort.get(); if (thread != null && !thread.equals(Thread.currentThread())) { logger.info("Shutdown ignored - container is stopping due to an aborted consumer"); return; } - try { - List canceledConsumers = new ArrayList<>(); - synchronized (this.consumersMonitor) { - if (this.consumers != null) { - Iterator consumerIterator = this.consumers.iterator(); - while (consumerIterator.hasNext()) { - BlockingQueueConsumer consumer = consumerIterator.next(); - consumer.basicCancel(true); - canceledConsumers.add(consumer); - consumerIterator.remove(); - if (consumer.declaring) { - consumer.thread.interrupt(); - } + List canceledConsumers = new ArrayList<>(); + synchronized (this.consumersMonitor) { + if (this.consumers != null) { + Iterator consumerIterator = this.consumers.iterator(); + while (consumerIterator.hasNext()) { + BlockingQueueConsumer consumer = consumerIterator.next(); + consumer.basicCancel(true); + canceledConsumers.add(consumer); + consumerIterator.remove(); + if (consumer.declaring) { + consumer.thread.interrupt(); } } - else { - logger.info("Shutdown ignored - container is already stopped"); - return; - } - } - logger.info("Waiting for workers to finish."); - boolean finished = this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS); - if (finished) { - logger.info("Successfully waited for workers to finish."); } else { - logger.info("Workers not finished."); - if (isForceCloseChannel()) { - canceledConsumers.forEach(consumer -> { - if (logger.isWarnEnabled()) { - logger.warn("Closing channel for unresponsive consumer: " + consumer); - } - consumer.stop(); - }); - } + logger.info("Shutdown ignored - container is already stopped"); + return; } } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.warn("Interrupted waiting for workers. Continuing with shutdown."); - } - synchronized (this.consumersMonitor) { - this.consumers = null; - this.cancellationLock.deactivate(); - } + Runnable awaitShutdown = () -> { + logger.info("Waiting for workers to finish."); + try { + boolean finished = this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS); + if (finished) { + logger.info("Successfully waited for workers to finish."); + } + else { + logger.info("Workers not finished."); + if (isForceCloseChannel()) { + canceledConsumers.forEach(consumer -> { + if (logger.isWarnEnabled()) { + logger.warn("Closing channel for unresponsive consumer: " + consumer); + } + consumer.stop(); + }); + } + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.warn("Interrupted waiting for workers. Continuing with shutdown."); + } + synchronized (this.consumersMonitor) { + this.consumers = null; + this.cancellationLock.deactivate(); + } + + if (callback != null) { + callback.run(); + } + }; + if (callback == null) { + awaitShutdown.run(); + } + else { + getTaskExecutor().execute(awaitShutdown); + } } private boolean isActive(BlockingQueueConsumer consumer) {