null if none found
*/
- public static RabbitResourceHolder getTransactionalResourceHolder(final ConnectionFactory cf,
+ public static RabbitResourceHolder getTransactionalResourceHolder(final ConnectionFactory connectionFactory,
final boolean synchedLocalTransactionAllowed) {
- return doGetTransactionalResourceHolder(cf, new ResourceFactory() {
- public Channel getChannel(RabbitResourceHolder holder) {
- return holder.getChannel();
- }
+ RabbitResourceHolder holder = doGetTransactionalResourceHolder(connectionFactory, new ResourceFactory() {
+ public Channel getChannel(RabbitResourceHolder holder) {
+ return holder.getChannel();
+ }
- public Connection getConnection(RabbitResourceHolder holder) {
- return holder.getConnection();
- }
+ public Connection getConnection(RabbitResourceHolder holder) {
+ return holder.getConnection();
+ }
- public Connection createConnection() throws IOException {
- return cf.createConnection();
- }
+ public Connection createConnection() throws IOException {
+ return connectionFactory.createConnection();
+ }
- public Channel createChannel(Connection con) throws IOException {
- return con.createChannel(synchedLocalTransactionAllowed);
- }
+ public Channel createChannel(Connection con) throws IOException {
+ return con.createChannel(synchedLocalTransactionAllowed);
+ }
- public boolean isSynchedLocalTransactionAllowed() {
- return synchedLocalTransactionAllowed;
- }
- });
+ public boolean isSynchedLocalTransactionAllowed() {
+ return synchedLocalTransactionAllowed;
+ }
+ });
+ if (synchedLocalTransactionAllowed) {
+ holder.declareTransactional();
+ }
+ return holder;
}
/**
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
index 9197fee4..8112dc75 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
@@ -12,6 +12,8 @@ import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.support.RabbitUtils;
import com.rabbitmq.client.AMQP;
@@ -23,9 +25,7 @@ import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.utility.Utility;
/**
- * Variation on QueueingConsumer in RabbitMQ, uses 'put' instead of 'add' and stored a reference to the consumerTag that
- * was returned when this Consumer was registered with the channel so as to make it easy to close the consumer when
- * shutting down.
+ * Specialized consumer encapsulating knowledge of the broker connections and having its own lifecycle (start and stop).
*
* @author Mark Pollack
* @author Dave Syer
@@ -35,6 +35,7 @@ public class BlockingQueueConsumer {
private static Log logger = LogFactory.getLog(BlockingQueueConsumer.class);
+ // This must be an unbounded queue or we risk blocking the Connection thread.
private final BlockingQueue- * Public setter for the {@link Advice} to apply to listener executions. If - * {@link #setTxSize(int) txSize>1} then multiple listener executions will - * all be wrapped in the same advice up to that limit. + * Public setter for the {@link Advice} to apply to listener executions. If {@link #setTxSize(int) txSize>1} then + * multiple listener executions will all be wrapped in the same advice up to that limit. *
*- * If a {@link #setTransactionManager(PlatformTransactionManager) - * transactionManager} is provided as well, then separate advice is created - * for the transaction and applied first in the chain. In that case the - * advice chain provided here should not contain a transaction interceptor - * (otherwise two transactions would be be applied). + * If a {@link #setTransactionManager(PlatformTransactionManager) transactionManager} is provided as well, then + * separate advice is created for the transaction and applied first in the chain. In that case the advice chain + * provided here should not contain a transaction interceptor (otherwise two transactions would be be applied). *
* - * @param advices - * the advice chain to set + * @param advices the advice chain to set */ public void setAdviceChain(Advice[] advices) { this.advices = advices; } + /** + * Specify the interval between recovery attempts, in milliseconds. The default is 5000 ms, that is, 5 + * seconds. + * @see #handleConsumerStartupFailure + */ + public void setRecoveryInterval(long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + public SimpleMessageListenerContainer() { } @@ -125,14 +133,12 @@ public class SimpleMessageListenerContainer extends /** * Specify the number of concurrent consumers to create. Default is 1. *
- * Raising the number of concurrent consumers is recommended in order to
- * scale the consumption of messages coming in from a queue. However, note
- * that any ordering guarantees are lost once multiple consumers are
- * registered. In general, stick with 1 consumer for low-volume queues.
+ * Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in
+ * from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In
+ * general, stick with 1 consumer for low-volume queues.
*/
public void setConcurrentConsumers(int concurrentConsumers) {
- Assert.isTrue(concurrentConsumers > 0,
- "'concurrentConsumers' value must be at least 1 (one)");
+ Assert.isTrue(concurrentConsumers > 0, "'concurrentConsumers' value must be at least 1 (one)");
this.concurrentConsumers = concurrentConsumers;
}
@@ -141,15 +147,12 @@ public class SimpleMessageListenerContainer extends
}
/**
- * The time to wait for workers in milliseconds after the container is
- * stopped, and before the connection is forced closed. If any workers are
- * active when the shutdown signal comes they will be allowed to finish
- * processing as long as they can finish within this timeout. Otherwise the
- * connection is closed and messages remain unacked (if the channel is
- * transactional). Defaults to 5 seconds.
+ * The time to wait for workers in milliseconds after the container is stopped, and before the connection is forced
+ * closed. If any workers are active when the shutdown signal comes they will be allowed to finish processing as
+ * long as they can finish within this timeout. Otherwise the connection is closed and messages remain unacked (if
+ * the channel is transactional). Defaults to 5 seconds.
*
- * @param shutdownTimeout
- * the shutdown timeout to set
+ * @param shutdownTimeout the shutdown timeout to set
*/
public void setShutdownTimeout(long shutdownTimeout) {
this.shutdownTimeout = shutdownTimeout;
@@ -161,47 +164,39 @@ public class SimpleMessageListenerContainer extends
}
/**
- * Tells the broker how many messages to send to each consumer in a single
- * request. Often this can be set quite high to improve throughput. It
- * should be greater than or equal to {@link #setTxSize(int) the transaction
- * size}.
+ * Tells the broker how many messages to send to each consumer in a single request. Often this can be set quite high
+ * to improve throughput. It should be greater than or equal to {@link #setTxSize(int) the transaction size}.
*
- * @param prefetchCount
- * the prefetch count
+ * @param prefetchCount the prefetch count
*/
public void setPrefetchCount(int prefetchCount) {
this.prefetchCount = prefetchCount;
}
/**
- * Tells the container how many messages to process in a single transaction
- * (if the channel is transactional). For best results it should be less
- * than or equal to {@link #setPrefetchCount(int) the prefetch count}.
+ * Tells the container how many messages to process in a single transaction (if the channel is transactional). For
+ * best results it should be less than or equal to {@link #setPrefetchCount(int) the prefetch count}.
*
- * @param prefetchCount
- * the prefetch count
+ * @param prefetchCount the prefetch count
*/
public void setTxSize(int txSize) {
this.txSize = txSize;
}
- public void setTransactionManager(
- PlatformTransactionManager transactionManager) {
+ public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
/**
- * @param transactionAttribute
- * the transaction attribute to set
+ * @param transactionAttribute the transaction attribute to set
*/
- public void setTransactionAttribute(
- TransactionAttribute transactionAttribute) {
+ public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
this.transactionAttribute = transactionAttribute;
}
/**
- * Avoid the possibility of not configuring the CachingConnectionFactory in
- * sync with the number of concurrent consumers.
+ * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent
+ * consumers.
*/
@Override
protected void validateConfiguration() {
@@ -239,7 +234,7 @@ public class SimpleMessageListenerContainer extends
}
}
- public void initializeProxy() {
+ private void initializeProxy() {
if (advices.length == 0 && transactionManager == null) {
return;
}
@@ -247,12 +242,10 @@ public class SimpleMessageListenerContainer extends
if (transactionManager != null) {
MatchAlwaysTransactionAttributeSource txAttributeSource = new MatchAlwaysTransactionAttributeSource();
txAttributeSource.setTransactionAttribute(transactionAttribute);
- Advice txAdvice = new TransactionInterceptor(transactionManager,
- txAttributeSource);
- factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE,
- txAdvice));
+ Advice txAdvice = new TransactionInterceptor(transactionManager, txAttributeSource);
+ factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, txAdvice));
}
- for (Advice advice : advices) {
+ for (Advice advice : getAdvices()) {
factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, advice));
}
factory.setProxyTargetClass(false);
@@ -273,8 +266,8 @@ public class SimpleMessageListenerContainer extends
}
/**
- * Creates the specified number of concurrent consumers, in the form of a
- * Rabbit Channel plus associated MessageConsumer.
+ * Creates the specified number of concurrent consumers, in the form of a Rabbit Channel plus associated
+ * MessageConsumer.
*
* @throws Exception
*/
@@ -286,16 +279,25 @@ public class SimpleMessageListenerContainer extends
return (int) cancellationLock.getCount();
}
+ @Override
+ protected void establishSharedConnection() throws Exception {
+ try {
+ super.establishSharedConnection();
+ } catch (AmqpException e) {
+ // Try to recover later...
+ logger.info("Could not start message listener container. Consumer threads will attempt to reconnect. Exception ("
+ + e.getClass().getName() + "): " + e.getMessage());
+ }
+ }
+
/**
- * Re-initializes this container's Rabbit message consumers, if not
- * initialized already. Then submits each consumer to this container's task
- * executor.
+ * Re-initializes this container's Rabbit message consumers, if not initialized already. Then submits each consumer
+ * to this container's task executor.
*
* @throws Exception
*/
protected void doStart() throws Exception {
super.doStart();
- establishSharedConnection();
initializeConsumers();
synchronized (this.consumersMonitor) {
if (this.consumers == null) {
@@ -304,8 +306,7 @@ public class SimpleMessageListenerContainer extends
}
cancellationLock = new CountDownLatch(this.consumers.size());
for (BlockingQueueConsumer consumer : this.consumers) {
- this.taskExecutor.execute(new AsyncMessageProcessingConsumer(
- consumer, cancellationLock));
+ this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, cancellationLock));
}
}
}
@@ -324,8 +325,7 @@ public class SimpleMessageListenerContainer extends
try {
logger.debug("Waiting for workers to finish.");
- boolean finished = cancellationLock.await(shutdownTimeout,
- TimeUnit.MILLISECONDS);
+ boolean finished = cancellationLock.await(shutdownTimeout, TimeUnit.MILLISECONDS);
if (finished) {
logger.info("Successfully waited for workers to finish.");
} else {
@@ -342,15 +342,12 @@ public class SimpleMessageListenerContainer extends
}
- protected void initializeConsumers() throws IOException {
+ protected void initializeConsumers() {
synchronized (this.consumersMonitor) {
if (this.consumers == null) {
- this.consumers = new HashSet