GH-3005: Fix SimpleMLC.killOrRestart for closed AC

Fixes: #3005
Issue link: #3005

The `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()`
is also called during application context shutdown.
At this moment we cannot emit events into an application context.
Otherwise, it fails with:
```
Exception in thread "rabbitListenerExecutor1" org.springframework.beans.factory.BeanCreationNotAllowedException: Error creating bean with name 'refreshEventListener': Singleton bean creation not allowed while singletons of this factory are in destruction (Do not request a bean from a BeanFactory in a destroy method implementation!)
```

* Introduce `ObservableListenerContainer.isApplicationContextClosed()`
and call it as additional condition in the `SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.killOrRestart()`
before trying to emit `AsyncConsumerStoppedEvent`

**Auto-cherry-pick to `3.2.x`**

The fix for `3.1.x` requires a slightly different approach via `ContextClosedEvent`

Author: artembilan

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/ObservableListenerContainer.java

@@ -26,6 +26,7 @@
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -121,6 +122,12 @@ protected void checkObservation() {
 		}
 	}
 
+
+	protected boolean isApplicationContextClosed() {
+		return this.applicationContext instanceof ConfigurableApplicationContext configurableCtx
+				&& configurableCtx.isClosed();
+	}
+
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;

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

@@ -64,6 +64,7 @@
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.amqp.support.ConsumerTagStrategy;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.log.LogMessage;
 import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.support.MetricType;
@@ -807,7 +808,6 @@ protected void adjustConsumers(int deltaArg) {
 		}
 	}
 
-
 	/**
 	 * Start up to delta consumers, limited by {@link #setMaxConcurrentConsumers(int)}.
 	 * @param delta the consumers to add.
@@ -875,7 +875,6 @@ private void considerAddingAConsumer() {
 		}
 	}
 
-
 	private void considerStoppingAConsumer(BlockingQueueConsumer consumer) {
 		this.consumersLock.lock();
 		try {
@@ -1272,7 +1271,6 @@ private final class AsyncMessageProcessingConsumer implements Runnable {
 
 		private boolean failedExclusive;
 
-
 		AsyncMessageProcessingConsumer(BlockingQueueConsumer consumer) {
 			this.consumer = consumer;
 			this.start = new CountDownLatch(1);
@@ -1556,8 +1554,9 @@ private void killOrRestart(boolean aborted) {
 				try {
 					this.consumer.stop();
 					SimpleMessageListenerContainer.this.cancellationLock.release(this.consumer);
-					if (getApplicationEventPublisher() != null) {
-						getApplicationEventPublisher().publishEvent(
+					ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
+					if (applicationEventPublisher != null && !isApplicationContextClosed()) {
+						applicationEventPublisher.publishEvent(
 								new AsyncConsumerStoppedEvent(SimpleMessageListenerContainer.this, this.consumer));
 					}
 				}