Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1436: Async Stop Containers

Browse Source 
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
This commit is contained in:
Mat Jaggard
2022-03-16 12:37:26 +00:00
committed by Gary Russell
parent d660edbfc1
commit 55d1289a35
2 changed files with 76 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -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();
}
}

105
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -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<BlockingQueueConsumer> canceledConsumers = new ArrayList<>();
synchronized (this.consumersMonitor) {
if (this.consumers != null) {
Iterator<BlockingQueueConsumer> 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<BlockingQueueConsumer> canceledConsumers = new ArrayList<>();
synchronized (this.consumersMonitor) {
if (this.consumers != null) {
Iterator<BlockingQueueConsumer> 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) {