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

GH-1401: SMLC: Fix setConcurrency

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1401

- multiple calls can temporarly inflate consumer count
This commit is contained in:
zysaaa
2021-12-14 02:24:41 +08:00
committed by Gary Russell
parent 3f003cfdfd
commit 07443ceb77
2 changed files with 27 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -223,7 +223,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
int maxConsumersToSet = Integer.parseInt(concurrency.substring(separatorIndex + 1));
Assert.isTrue(maxConsumersToSet >= consumersToSet,
"'maxConcurrentConsumers' value must be at least 'concurrentConsumers'");
this.concurrentConsumers = 1;
this.maxConcurrentConsumers = null;
setConcurrentConsumers(consumersToSet);
setMaxConcurrentConsumers(maxConsumersToSet);

28
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
View File

@@ -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.
@@ -69,6 +69,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -650,6 +651,31 @@ public class SimpleMessageListenerContainerTests {
assertThat(afterReceivePostProcessors).containsExactly(mpp2, mpp3);
}
@Test
void setConcurrency() throws Exception {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
Connection connection = mock(Connection.class);
Channel channel = mock(Channel.class);
given(connectionFactory.createConnection()).willReturn(connection);
given(connection.createChannel(false)).willReturn(channel);
final AtomicReference<Consumer> consumer = new AtomicReference<>();
willAnswer(invocation -> {
consumer.set(invocation.getArgument(6));
consumer.get().handleConsumeOk("1");
return "1";
}).given(channel)
.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(),
any(Consumer.class));
final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setQueueNames("foo", "bar");
container.setMessageListener(mock(MessageListener.class));
container.setConcurrency("5-10");
container.start();
await().until(() -> TestUtils.getPropertyValue(container, "consumers", Collection.class).size() == 5);
container.setConcurrency("10-10");
assertThat(TestUtils.getPropertyValue(container, "consumers", Collection.class)).hasSize(10);
}
private Answer<Object> messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container,
final boolean cancel, final CountDownLatch latch) {
return invocation -> {