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

AMQP-223 Fix Consumer Thread Management

Browse Source 
If doStart() was called multiple times, multiple threads ran
in each consumer.

There is a check to prevent creating multiple consumers in
this case, but the thread management had no such check.
This commit is contained in:
Gary Russell
2012-04-06 11:37:27 -04:00
committed by Oleg Zhurakousky
parent c4170ce40f
commit 628707ee32
1 changed files with 14 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -292,9 +292,17 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
protected void doStart() throws Exception {
super.doStart();
synchronized (this.consumersMonitor) {
initializeConsumers();
int newConsumers = initializeConsumers();
if (this.consumers == null) {
logger.info("Consumers were initialized and then cleared (presumably the container was stopped concurrently)");
if (logger.isInfoEnabled()) {
logger.info("Consumers were initialized and then cleared (presumably the container was stopped concurrently)");
}
return;
}
if (newConsumers <= 0) {
if (logger.isInfoEnabled()) {
logger.info("Consumers are already running");
}
return;
}
Set<AsyncMessageProcessingConsumer> processors = new HashSet<AsyncMessageProcessingConsumer>();
@@ -343,7 +351,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
protected void initializeConsumers() {
protected int initializeConsumers() {
int count = 0;
synchronized (this.consumersMonitor) {
if (this.consumers == null) {
cancellationLock.reset();
@@ -351,9 +360,11 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
for (int i = 0; i < this.concurrentConsumers; i++) {
BlockingQueueConsumer consumer = createBlockingQueueConsumer();
this.consumers.add(consumer);
count++;
}
}
}
return count;
}
protected boolean isChannelLocallyTransacted(Channel channel) {