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

AMQP-810: Fix adjust consumers when max present

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-810

SMLC: adjusting the `concurrentConsumers` did not consider `maxConcurrentConsumers`.
- increase added consumers even if at max
- decrease removed consumers when they had increased due to max being set

Further, decreasing the `maxConcurrentConsumers` did not remove consumers if there
were more consumers than the new max.

- don't add consumers beyond the max
- don't remove consumers unless the new max is exceeded

**cherry-pick to 2.0.x, 1.7.x**

There will be some minor conflicts in 1.7.x since the modified test is JUnit5.
This commit is contained in:
Gary Russell
2018-04-27 11:15:45 -04:00
committed by Artem Bilan
parent f678078e29
commit daebf40a69
2 changed files with 144 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -164,19 +164,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
int delta = this.concurrentConsumers - concurrentConsumers;
this.concurrentConsumers = concurrentConsumers;
if (isActive() && this.consumers != null) {
if (delta > 0) {
Iterator<BlockingQueueConsumer> consumerIterator = this.consumers.iterator();
while (consumerIterator.hasNext() && delta > 0) {
BlockingQueueConsumer consumer = consumerIterator.next();
consumer.basicCancel(true);
consumerIterator.remove();
delta--;
}
}
else {
addAndStartConsumers(-delta);
}
if (isActive()) {
adjustConsumers(delta);
}
}
}
@@ -196,7 +185,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
"'maxConcurrentConsumers' value must be at least 'concurrentConsumers'");
Assert.isTrue(!isExclusive() || maxConcurrentConsumers == 1,
"When the consumer is exclusive, the concurrency must be 1");
Integer oldMax = this.maxConcurrentConsumers;
this.maxConcurrentConsumers = maxConcurrentConsumers;
if (oldMax != null && isActive()) {
int delta = oldMax - maxConcurrentConsumers;
if (delta > 0) { // only decrease, not increase
adjustConsumers(delta);
}
}
}
/**
@@ -587,10 +584,45 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
return count;
}
/**
* Adjust consumers depending on delta.
* @param delta a negative value increases, positive decreases.
* @since 1.7.8
*/
protected void adjustConsumers(int delta) {
synchronized (this.consumersMonitor) {
if (isActive() && this.consumers != null) {
if (delta > 0) {
Iterator<BlockingQueueConsumer> consumerIterator = this.consumers.iterator();
while (consumerIterator.hasNext() && delta > 0
&& (this.maxConcurrentConsumers == null
|| this.consumers.size() > this.maxConcurrentConsumers)) {
BlockingQueueConsumer consumer = consumerIterator.next();
consumer.basicCancel(true);
consumerIterator.remove();
delta--;
}
}
else {
addAndStartConsumers(-delta);
}
}
}
}
/**
* Start up to delta consumers, limited by {@link #setMaxConcurrentConsumers(int)}.
* @param delta the consumers to add.
*/
protected void addAndStartConsumers(int delta) {
synchronized (this.consumersMonitor) {
if (this.consumers != null) {
for (int i = 0; i < delta; i++) {
if (this.maxConcurrentConsumers != null
&& this.consumers.size() >= this.maxConcurrentConsumers) {
break;
}
BlockingQueueConsumer consumer = createBlockingQueueConsumer();
this.consumers.add(consumer);
AsyncMessageProcessingConsumer processor = new AsyncMessageProcessingConsumer(consumer);

101
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -16,7 +16,9 @@
package org.springframework.amqp.rabbit.listener;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import java.util.Set;
@@ -44,12 +46,23 @@ import com.rabbitmq.client.ConnectionFactory;
* @since 1.2.1
*
*/
@RabbitAvailable(queues = SimpleMessageListenerContainerLongTests.QUEUE)
@RabbitAvailable(queues = {
SimpleMessageListenerContainerLongTests.QUEUE,
SimpleMessageListenerContainerLongTests.QUEUE2,
SimpleMessageListenerContainerLongTests.QUEUE3,
SimpleMessageListenerContainerLongTests.QUEUE4
})
@LongRunning
public class SimpleMessageListenerContainerLongTests {
public static final String QUEUE = "SimpleMessageListenerContainerLongTests.queue";
public static final String QUEUE2 = "SimpleMessageListenerContainerLongTests.queue2";
public static final String QUEUE3 = "SimpleMessageListenerContainerLongTests.queue3";
public static final String QUEUE4 = "SimpleMessageListenerContainerLongTests.queue4";
private final Log logger = LogFactory.getLog(SimpleMessageListenerContainerLongTests.class);
private final SingleConnectionFactory connectionFactory;
@@ -132,6 +145,90 @@ public class SimpleMessageListenerContainerLongTests {
connectionFactory.destroy();
}
@Test
public void testIncreaseMinAtMax() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setStartConsumerMinInterval(100);
container.setConsecutiveActiveTrigger(1);
container.setMessageListener(m -> {
try {
Thread.sleep(50);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
container.setQueueNames(QUEUE2);
container.setConcurrentConsumers(2);
container.setMaxConcurrentConsumers(5);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
for (int i = 0; i < 20; i++) {
template.convertAndSend(QUEUE2, "foo");
}
waitForNConsumers(container, 5);
container.setConcurrentConsumers(4);
Set<?> consumers = (Set<?>) TestUtils.getPropertyValue(container, "consumers");
assertThat(consumers.size(), equalTo(5));
}
@Test
public void testDecreaseMinAtMax() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setStartConsumerMinInterval(100);
container.setConsecutiveActiveTrigger(1);
container.setMessageListener(m -> {
try {
Thread.sleep(50);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
container.setQueueNames(QUEUE3);
container.setConcurrentConsumers(2);
container.setMaxConcurrentConsumers(3);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
for (int i = 0; i < 20; i++) {
template.convertAndSend(QUEUE3, "foo");
}
waitForNConsumers(container, 3);
container.setConcurrentConsumers(1);
Set<?> consumers = (Set<?>) TestUtils.getPropertyValue(container, "consumers");
assertThat(consumers.size(), equalTo(3));
}
@Test
public void testDecreaseMaxAtMax() throws Exception {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactory);
container.setStartConsumerMinInterval(100);
container.setConsecutiveActiveTrigger(1);
container.setMessageListener(m -> {
try {
Thread.sleep(50);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
container.setQueueNames(QUEUE4);
container.setConcurrentConsumers(2);
container.setMaxConcurrentConsumers(3);
container.afterPropertiesSet();
container.start();
RabbitTemplate template = new RabbitTemplate(this.connectionFactory);
for (int i = 0; i < 20; i++) {
template.convertAndSend(QUEUE4, "foo");
}
waitForNConsumers(container, 3);
container.setConcurrentConsumers(1);
container.setMaxConcurrentConsumers(1);
Set<?> consumers = (Set<?>) TestUtils.getPropertyValue(container, "consumers");
assertThat(consumers.size(), equalTo(1));
}
public void handleMessage(String foo) {
logger.info(foo);