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 7b44f625..ec35ee24 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-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -1757,6 +1757,13 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } } + protected void publishMissingQueueEvent(String queue) { + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher + .publishEvent(new MissingQueueEvent(this, queue)); + } + } + protected final void publishIdleContainerEvent(long idleTime) { if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 8619da1c..fc2d6fa7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -160,6 +160,8 @@ public class BlockingQueueConsumer { private long consumeDelay; + private java.util.function.Consumer missingQueuePublisher = str -> { }; + private volatile long abortStarted; private volatile boolean normalCancel; @@ -374,6 +376,15 @@ public class BlockingQueueConsumer { this.applicationEventPublisher = applicationEventPublisher; } + /** + * Set the publisher for a missing queue event. + * @param missingQueuePublisher the publisher. + * @since 2.1.18 + */ + public void setMissingQueuePublisher(java.util.function.Consumer missingQueuePublisher) { + this.missingQueuePublisher = missingQueuePublisher; + } + /** * Set the consumeDelay - a time to wait before consuming in ms. This is useful when * using the sharding plugin with {@code concurrency > 1}, to avoid uneven distribution of @@ -714,6 +725,7 @@ public class BlockingQueueConsumer { if (logger.isWarnEnabled()) { logger.warn("Failed to declare queue: " + queueName); } + this.missingQueuePublisher.accept(queueName); if (!this.channel.isOpen()) { throw new AmqpIOException(e); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index 23879ea9..03444813 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -24,12 +24,14 @@ import java.util.Collection; import java.util.Collections; import java.util.Date; import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Properties; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -103,7 +105,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta protected final List consumers = new LinkedList<>(); // NOSONAR - private final List consumersToRestart = new LinkedList<>(); + private final Set consumersToRestart = new LinkedHashSet<>(); + + private final Set removedQueues = ConcurrentHashMap.newKeySet(); private final MultiValueMap consumersByQueue = new LinkedMultiValueMap<>(); @@ -243,6 +247,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta Assert.notNull(queueNames, "'queueNames' cannot be null"); Assert.noNullElements(queueNames, "'queueNames' cannot contain null elements"); try { + Arrays.stream(queueNames).forEach(this.removedQueues::remove); addQueues(Arrays.stream(queueNames)); } catch (AmqpIOException e) { @@ -256,6 +261,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta Assert.notNull(queues, "'queues' cannot be null"); Assert.noNullElements(queues, "'queues' cannot contain null elements"); try { + Arrays.stream(queues) + .map(q -> q.getActualName()) + .forEach(this.removedQueues::remove); addQueues(Arrays.stream(queues).map(Queue::getName)); } catch (AmqpIOException e) { @@ -298,7 +306,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta if (isRunning()) { synchronized (this.consumersMonitor) { checkStartState(); - queueNames.map(this.consumersByQueue::remove) + queueNames.map(queue -> { + this.removedQueues.add(queue); + return this.consumersByQueue.remove(queue); + }) .filter(Objects::nonNull) .flatMap(Collection::stream) .forEach(this::cancelConsumer); @@ -313,7 +324,21 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta for (String queue : getQueueNames()) { while (this.consumersByQueue.get(queue) == null || this.consumersByQueue.get(queue).size() < newCount) { // NOSONAR never null - doConsumeFromQueue(queue); + List cBQ = this.consumersByQueue.get(queue); + int index = 0; + if (cBQ != null) { + // find a gap or set the index to the end + List indices = cBQ.stream() + .map(cons -> cons.getIndex()) + .sorted() + .collect(Collectors.toList()); + for (index = 0; index < indices.size(); index++) { + if (index < indices.get(index)) { + break; + } + } + } + doConsumeFromQueue(queue, index); } List consumerList = this.consumersByQueue.get(queue); if (consumerList != null && consumerList.size() > newCount) { @@ -430,9 +455,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta checkConsumers(now); if (this.lastRestartAttempt + getFailedDeclarationRetryInterval() < now) { synchronized (this.consumersMonitor) { - List restartableConsumers = new ArrayList<>(this.consumersToRestart); - this.consumersToRestart.clear(); if (this.started) { + List restartableConsumers = new ArrayList<>(this.consumersToRestart); + this.consumersToRestart.clear(); if (restartableConsumers.size() > 0) { doRedeclareElementsIfNecessary(); } @@ -440,10 +465,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta while (iterator.hasNext()) { SimpleConsumer consumer = iterator.next(); iterator.remove(); - if (!DirectMessageListenerContainer.this.consumersByQueue - .containsKey(consumer.getQueue())) { + if (DirectMessageListenerContainer.this.removedQueues.contains(consumer.getQueue())) { if (this.logger.isDebugEnabled()) { - this.logger.debug("Skipping restart of consumer " + consumer); + this.logger.debug("Skipping restart of consumer, queue removed " + consumer); } continue; } @@ -516,11 +540,11 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta if (StringUtils.hasText(actualName)) { namesToQueues.remove(consumer.getQueue()); namesToQueues.put(actualName, queue); - consumer = new SimpleConsumer(null, null, actualName); + consumer = new SimpleConsumer(null, null, actualName, consumer.getIndex()); } } try { - doConsumeFromQueue(consumer.getQueue()); + doConsumeFromQueue(consumer.getQueue(), consumer.getIndex()); return true; } catch (AmqpConnectException | AmqpIOException e) { @@ -646,12 +670,12 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta // Possible race with setConsumersPerQueue and the task launched by start() if (CollectionUtils.isEmpty(list)) { for (int i = 0; i < this.consumersPerQueue; i++) { - doConsumeFromQueue(queue); + doConsumeFromQueue(queue, i); } } } - private void doConsumeFromQueue(String queue) { + private void doConsumeFromQueue(String queue, int index) { if (!isActive()) { if (this.logger.isDebugEnabled()) { this.logger.debug("Consume from queue " + queue + " ignore, container stopping"); @@ -668,7 +692,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } catch (Exception e) { publishConsumerFailedEvent(e.getMessage(), false, e); - addConsumerToRestart(new SimpleConsumer(null, null, queue)); + addConsumerToRestart(new SimpleConsumer(null, null, queue, index)); throw e instanceof AmqpConnectException // NOSONAR exception type check ? (AmqpConnectException) e : new AmqpConnectException(e); @@ -678,7 +702,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta SimpleResourceHolder.pop(getRoutingConnectionFactory()); // NOSONAR never null here } } - SimpleConsumer consumer = consume(queue, connection); + SimpleConsumer consumer = consume(queue, index, connection); synchronized (this.consumersMonitor) { if (consumer != null) { this.cancellationLock.add(consumer); @@ -695,7 +719,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } @Nullable - private SimpleConsumer consume(String queue, Connection connection) { + private SimpleConsumer consume(String queue, int index, Connection connection) { Channel channel = null; SimpleConsumer consumer = null; try { @@ -709,7 +733,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } channel = connection.createChannel(isChannelTransacted()); channel.basicQos(getPrefetchCount()); - consumer = new SimpleConsumer(connection, channel, queue); + consumer = new SimpleConsumer(connection, channel, queue, index); channel.queueDeclarePassive(queue); consumer.consumerTag = channel.basicConsume(queue, getAcknowledgeMode().isAutoAck(), (getConsumerTagStrategy() != null @@ -723,13 +747,14 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta RabbitUtils.closeChannel(channel); RabbitUtils.closeConnection(connection); - consumer = handleConsumeException(queue, consumer, e); + consumer = handleConsumeException(queue, index, consumer, e); } return consumer; } @Nullable - private SimpleConsumer handleConsumeException(String queue, @Nullable SimpleConsumer consumerArg, Exception e) { + private SimpleConsumer handleConsumeException(String queue, int index, @Nullable SimpleConsumer consumerArg, + Exception e) { SimpleConsumer consumer = consumerArg; if (e.getCause() instanceof ShutdownSignalException @@ -740,6 +765,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } else if (e.getCause() instanceof ShutdownSignalException && RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) e.getCause())) { + publishMissingQueueEvent(queue); this.logger.error("Queue not present, scheduling consumer " + (consumer == null ? "for queue " + queue : consumer) + " for restart", e); } @@ -749,7 +775,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } if (consumer == null) { - addConsumerToRestart(new SimpleConsumer(null, null, queue)); + addConsumerToRestart(new SimpleConsumer(null, null, queue, index)); } else { addConsumerToRestart(consumer); @@ -844,11 +870,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } private void addConsumerToRestart(SimpleConsumer consumer) { - if (this.started) { - this.consumersToRestart.add(consumer); - if (this.logger.isTraceEnabled()) { - this.logger.trace("Consumers to restart now: " + this.consumersToRestart); - } + this.consumersToRestart.add(consumer); + if (this.logger.isTraceEnabled()) { + this.logger.trace("Consumers to restart now: " + this.consumersToRestart); } } @@ -871,6 +895,8 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta private final String queue; + private final int index; + private final boolean ackRequired; private final ConnectionFactory connectionFactory = getConnectionFactory(); @@ -903,10 +929,11 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta private volatile boolean ackFailed; - SimpleConsumer(@Nullable Connection connection, @Nullable Channel channel, String queue) { + SimpleConsumer(@Nullable Connection connection, @Nullable Channel channel, String queue, int index) { super(channel); this.connection = connection; this.queue = queue; + this.index = index; this.ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual(); if (channel instanceof ChannelProxy) { this.targetChannel = ((ChannelProxy) channel).getTargetChannel(); @@ -916,10 +943,14 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } } - private String getQueue() { + String getQueue() { return this.queue; } + int getIndex() { + return this.index; + } + @Override public String getConsumerTag() { return this.consumerTag; @@ -1220,9 +1251,53 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta consumerRemoved(this); } + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + getEnclosingInstance().hashCode(); + result = prime * result + this.index; + result = prime * result + ((this.queue == null) ? 0 : this.queue.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + SimpleConsumer other = (SimpleConsumer) obj; + if (!getEnclosingInstance().equals(other.getEnclosingInstance())) { + return false; + } + if (this.index != other.index) { + return false; + } + if (this.queue == null) { + if (other.queue != null) { + return false; + } + } + else if (!this.queue.equals(other.queue)) { + return false; + } + return true; + } + + private DirectMessageListenerContainer getEnclosingInstance() { + return DirectMessageListenerContainer.this; + } + @Override public String toString() { - return "SimpleConsumer [queue=" + this.queue + ", consumerTag=" + this.consumerTag + return "SimpleConsumer [queue=" + this.queue + ", index=" + this.index + + ", consumerTag=" + this.consumerTag + " identity=" + ObjectUtils.getIdentityHexString(this) + "]"; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java new file mode 100644 index 00000000..e0243787 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java @@ -0,0 +1,57 @@ +/* + * Copyright 2021 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.listener; + +import org.springframework.amqp.event.AmqpEvent; + +/** + * Event published when a missing queue is detected. + * + * @author Gary Russell + * @since 2.1.18 + * + */ +public class MissingQueueEvent extends AmqpEvent { + + private static final long serialVersionUID = 1L; + + private final String queue; + + /** + * Construct an instance with the provided source and queue. + * @param source the source. + * @param queue the queue. + */ + public MissingQueueEvent(Object source, String queue) { + super(source); + this.queue = queue; + } + + /** + * Return the missing queue. + * @return the queue. + */ + public String getQueue() { + return this.queue; + } + + @Override + public String toString() { + return "MissingQueueEvent [queue=" + this.queue + ", source=" + this.source + "]"; + } + +} 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 d9a10a42..a67751c0 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-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -822,6 +822,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta consumer = new BlockingQueueConsumer(getConnectionFactory(), getMessagePropertiesConverter(), this.cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount, isDefaultRequeueRejected(), getConsumerArguments(), isNoLocal(), isExclusive(), queues); + consumer.setMissingQueuePublisher(this::publishMissingQueueEvent); if (this.declarationRetries != null) { consumer.setDeclarationRetries(this.declarationRetries); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index d28ab00e..5a2faf0a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -30,6 +30,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -101,6 +102,8 @@ public class DirectMessageListenerContainerIntegrationTests { public static final String DLQ1 = "testDLQ1.DirectMessageListenerContainerIntegrationTests"; + private static final String MISSING = "missing.DirectMessageListenerContainerIntegrationTests"; + private static CachingConnectionFactory adminCf; private static RabbitAdmin admin; @@ -679,6 +682,31 @@ public class DirectMessageListenerContainerIntegrationTests { cf.destroy(); } + @Test + void missingQueueOnStart() throws InterruptedException { + CachingConnectionFactory cf = new CachingConnectionFactory("localhost"); + RabbitAdmin admin = new RabbitAdmin(cf); + admin.deleteQueue(MISSING); + DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf); + container.setQueueNames(MISSING); + container.setBeanName("missingQOnStart"); + final CountDownLatch latch = new CountDownLatch(1); + container.setApplicationEventPublisher(event -> { + if (event instanceof MissingQueueEvent) { + admin.declareQueue(new Queue(MISSING)); + } + else if (event instanceof AsyncConsumerStartedEvent) { + latch.countDown(); + } + }); + container.afterPropertiesSet(); + container.start(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + container.stop(); + admin.deleteQueue(MISSING); + cf.destroy(); + } + private boolean consumersOnQueue(String queue, int expected) throws Exception { await().with().pollDelay(Duration.ZERO).atMost(Duration.ofSeconds(60)) .until(() -> admin.getQueueProperties(queue), @@ -694,7 +722,7 @@ public class DirectMessageListenerContainerIntegrationTests { } private boolean restartConsumerCount(AbstractMessageListenerContainer container, int expected) throws Exception { - List consumers = TestUtils.getPropertyValue(container, "consumersToRestart", List.class); + Set consumers = TestUtils.getPropertyValue(container, "consumersToRestart", Set.class); await().with().pollDelay(Duration.ZERO).atMost(Duration.ofSeconds(60)) .until(() -> consumers.size() == expected); return true; diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 4b42e510..84c7afee 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -241,8 +241,10 @@ public class SimpleMessageListenerContainerIntegration2Tests { if (event instanceof ListenerContainerConsumerFailedEvent) { eventRef.set((ListenerContainerConsumerFailedEvent) event); } - events.add((AmqpEvent) event); - eventLatch.countDown(); + if (!(event instanceof MissingQueueEvent)) { + events.add((AmqpEvent) event); + eventLatch.countDown(); + } }); container.start(); for (int i = 0; i < 10; i++) { @@ -518,6 +520,7 @@ public class SimpleMessageListenerContainerIntegration2Tests { ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort()); CountDownLatch latch = new CountDownLatch(1); + CountDownLatch missingLatch = new CountDownLatch(1); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch))); container.setQueues(queue); @@ -526,9 +529,16 @@ public class SimpleMessageListenerContainerIntegration2Tests { container.setDeclarationRetries(1); container.setFailedDeclarationRetryInterval(100); container.setRetryDeclarationInterval(30000); + container.setApplicationEventPublisher(event -> { + if (event instanceof MissingQueueEvent) { + missingLatch.countDown(); + } + }); container.afterPropertiesSet(); container.start(); + assertThat(missingLatch.await(10, TimeUnit.SECONDS)).isTrue(); + new RabbitAdmin(connectionFactory).declareQueue(queue); this.template.convertAndSend(queue.getName(), "foo"); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 41116aa2..d322a8a5 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -2056,6 +2056,7 @@ Several other events are published at various stages of the container lifecycle: * `AsyncConsumerStoppedEvent`: When the consumer is stopped - `SimpleMessageListenerContainer` only. * `ConsumeOkEvent`: When a `consumeOk` is received from the broker, contains the queue name and `consumerTag` * `ListenerContainerIdleEvent`: See <>. +* `MissingQueueEvent`: When a missing queue is detected. [[consumerTags]] ===== Consumer Tags