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

Merge pull request #23 from garyrussell/AMQP-223

Browse Source 
* AMQP-223:
  AMQP-223 Fix Consumer Thread Management
This commit is contained in:
Oleg Zhurakousky
2012-04-09 13:46:21 -04:00
parent c4170ce40f 628707ee32
commit 4a2e08ef4e
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) {