From e190e53dfd51450861400bd2c0f8ec8ef2ee0a3e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 7 Apr 2021 17:01:41 -0400 Subject: [PATCH] GH-1318: Support the Global Flag in basicQos Resolves https://github.com/spring-projects/spring-amqp/issues/1318 **cherry-pick to 2.2.x** --- .../AbstractRabbitListenerContainerFactory.java | 15 ++++++++++++++- .../config/ListenerContainerFactoryBean.java | 17 ++++++++++++++++- .../rabbit/config/RabbitNamespaceUtils.java | 9 ++++++++- .../AbstractMessageListenerContainer.java | 17 +++++++++++++++++ .../rabbit/listener/BlockingQueueConsumer.java | 16 +++++++++++++--- .../DirectMessageListenerContainer.java | 2 +- .../SimpleMessageListenerContainer.java | 1 + .../amqp/rabbit/config/spring-rabbit.xsd | 17 +++++++++++++++-- .../config/ListenerContainerParserTests.java | 4 +++- .../RabbitListenerContainerFactoryTests.java | 4 +++- .../listener/BlockingQueueConsumerTests.java | 8 ++++---- ...DirectMessageListenerContainerMockTests.java | 8 ++++---- ...ssageListenerContainerIntegration2Tests.java | 9 +++++++-- .../ListenerContainerParserTests-context.xml | 2 +- src/reference/asciidoc/amqp.adoc | 10 ++++++++++ src/reference/asciidoc/whats-new.adoc | 2 ++ 16 files changed, 119 insertions(+), 22 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java index accedb0d..ce3b05e4 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-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. @@ -86,6 +86,8 @@ public abstract class AbstractRabbitListenerContainerFactory missingQueuePublisher = str -> { }; + private boolean globalQos; + private volatile long abortStarted; private volatile boolean normalCancel; @@ -405,6 +407,16 @@ public class BlockingQueueConsumer { this.deliveryTags.clear(); } + /** + * Apply prefetch to the entire channel. + * @param globalQos true for a channel-wide prefetch. + * @since 2.2.17 + * @see Channel#basicQos(int, boolean) + */ + public void setGlobalQos(boolean globalQos) { + this.globalQos = globalQos; + } + /** * Return true if cancellation is expected. * @return true if expected. @@ -629,10 +641,8 @@ public class BlockingQueueConsumer { } } if (!this.acknowledgeMode.isAutoAck() && !cancelled()) { - // Set basicQos before calling basicConsume (otherwise if we are not acking the broker - // will send blocks of 100 messages) try { - this.channel.basicQos(this.prefetchCount); + this.channel.basicQos(this.prefetchCount, this.globalQos); } catch (IOException e) { this.activeObjectCounter.release(this); 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 86a7c7de..b47d043f 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 @@ -736,7 +736,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } } channel = connection.createChannel(isChannelTransacted()); - channel.basicQos(getPrefetchCount()); + channel.basicQos(getPrefetchCount(), isGlobalQos()); consumer = new SimpleConsumer(connection, channel, queue, index); channel.queueDeclarePassive(queue); consumer.consumerTag = channel.basicConsume(queue, getAcknowledgeMode().isAutoAck(), 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 a67751c0..0d75ef5b 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 @@ -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.setGlobalQos(isGlobalQos()); consumer.setMissingQueuePublisher(this::publishMissingQueueEvent); if (this.declarationRetries != null) { consumer.setDeclarationRetries(this.declarationRetries); diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd index bb308496..09bdec3b 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd @@ -728,11 +728,24 @@ + + + + + + + + c.setShutdownTimeout(10_000)); assertThat(this.factory.getAdviceChain()).isEqualTo(new Advice[]{advice}); @@ -138,6 +139,7 @@ public class RabbitListenerContainerFactoryTests { List actualAfterReceivePostProcessors = (List) fieldAccessor.getPropertyValue("afterReceivePostProcessors"); assertThat(actualAfterReceivePostProcessors.size()).as("Wrong number of afterReceivePostProcessors").isEqualTo(1); assertThat(actualAfterReceivePostProcessors.get(0)).as("Wrong advice").isSameAs(afterReceivePostProcessor); + assertThat(fieldAccessor.getPropertyValue("globalQos")).isEqualTo(true); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java index cd338855..17c9d893 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.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. @@ -161,7 +161,7 @@ public class BlockingQueueConsumerTests { blockingQueueConsumer.setFailedDeclarationRetryInterval(10); blockingQueueConsumer.start(); - verify(channel).basicQos(20); + verify(channel).basicQos(20, false); } @Test @@ -291,7 +291,7 @@ public class BlockingQueueConsumerTests { blockingQueueConsumer.setFailedDeclarationRetryInterval(10); blockingQueueConsumer.start(); - verify(channel).basicQos(2); + verify(channel).basicQos(2, false); isOpen.set(false); blockingQueueConsumer.stop(); verify(channel).basicCancel("consumerTag"); @@ -333,7 +333,7 @@ public class BlockingQueueConsumerTests { blockingQueueConsumer.setFailedDeclarationRetryInterval(10); blockingQueueConsumer.start(); - verify(channel).basicQos(2); + verify(channel).basicQos(2, false); Consumer consumer = (Consumer) TestUtils.getPropertyValue(blockingQueueConsumer, "consumers", Map.class) .get("test"); isOpen.set(false); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java index 4a8d902c..6d5dd3b8 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-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. @@ -86,7 +86,7 @@ public class DirectMessageListenerContainerMockTests { qos.set(i.getArgument(0)); latch1.countDown(); return null; - }).given(channel).basicQos(anyInt()); + }).given(channel).basicQos(anyInt(), anyBoolean()); final CountDownLatch latch2 = new CountDownLatch(1); willAnswer(i -> { latch2.countDown(); @@ -135,7 +135,7 @@ public class DirectMessageListenerContainerMockTests { willAnswer(i -> { qos.set(i.getArgument(0)); return null; - }).given(channel).basicQos(anyInt()); + }).given(channel).basicQos(anyInt(), anyBoolean()); final CountDownLatch latch2 = new CountDownLatch(2); final CountDownLatch latch3 = new CountDownLatch(1); willAnswer(i -> { @@ -233,7 +233,7 @@ public class DirectMessageListenerContainerMockTests { qos.set(i.getArgument(0)); latch1.countDown(); return null; - }).given(channel).basicQos(anyInt()); + }).given(channel).basicQos(anyInt(), anyBoolean()); final CountDownLatch latch2 = new CountDownLatch(2); willAnswer(i -> { latch2.countDown(); 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 84c7afee..42ccdc01 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 @@ -424,6 +424,8 @@ public class SimpleMessageListenerContainerIntegration2Tests { final AtomicBoolean networkGlitch = new AtomicBoolean(); + final AtomicBoolean globalQos = new AtomicBoolean(); + class MockChannel extends PublisherCallbackChannelImpl { MockChannel(Channel delegate) { @@ -431,11 +433,12 @@ public class SimpleMessageListenerContainerIntegration2Tests { } @Override - public void basicQos(int prefetchCount) throws IOException { + public void basicQos(int prefetchCount, boolean global) throws IOException { + globalQos.set(global); if (networkGlitch.compareAndSet(false, true)) { throw new IOException("Intentional connection reset"); } - super.basicQos(prefetchCount); + super.basicQos(prefetchCount, global); } } @@ -452,11 +455,13 @@ public class SimpleMessageListenerContainerIntegration2Tests { container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch))); container.setQueueNames(queue.getName()); container.setRecoveryInterval(500); + container.setGlobalQos(true); container.afterPropertiesSet(); container.start(); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(networkGlitch.get()).isTrue(); + assertThat(globalQos.get()).isTrue(); container.stop(); ((DisposableBean) connectionFactory).destroy(); diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml index 4a126685..30ccde87 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml @@ -24,7 +24,7 @@ + auto-declare="false" global-qos="true"> diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 88b2344f..c95761d1 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -5672,6 +5672,15 @@ You can set it to `false` to revert to the previous behavior. a|image::images/tickmark.png[] a|image::images/tickmark.png[] +|globalQos +(global-qos) + +|When true, the `prefetchCount` is applied globally to the channel rather than to each consumer on the channel. +See https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.qos.global[`basicQos.global`] for more information. + +a|image::images/tickmark.png[] +a|image::images/tickmark.png[] + |(group) |This is available only when using the namespace. @@ -5866,6 +5875,7 @@ to a large amount of memory in the client process), and if strict message orderi (the prefetch value should be set back to 1 in this case). Also, with low-volume messaging and multiple consumers (including concurrency within a single listener container instance), you may wish to reduce the prefetch to get a more even distribution of messages across consumers. +Also see `globalQos`. a|image::images/tickmark.png[] a|image::images/tickmark.png[] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 40099e71..5b53dc3b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -41,6 +41,8 @@ A new listener container property `consumeDelay` is now available; it is helpful The default `JavaLangErrorHandler` now calls `System.exit(99)`. To revert to the previous behavior (do nothing), add a no-op handler. +The containers now support the `globalQos` property to apply the `prefetchCount` globally for the channel rather than for each consumer on the channel. + See <> for more information. ==== MessagePostProcessor Changes