Spring AMQP

The minimum Spring Framework version dependency is 5.2.0.

The minimum amqp-client Java client library version is 5.7.0.

This section offers the fastest introduction.

First, add the following import statements to make the examples later in this section work:

The following example uses plain, imperative Java to send and receive a message:

Note that there is also a ConnectionFactory in the native Java Rabbit client. We use the Spring abstraction in the preceding code. It caches channels (and optionally connections) for reuse. We rely on the default exchange in the broker (since none is specified in the send), and the default binding of all queues to the default exchange by their name (thus, we can use the queue name as a routing key in the send). Those behaviors are defined in the AMQP specification.

The following example is the same as the preceding example but externalizes the resource configuration to XML:

By default, the <rabbit:admin/> declaration automatically looks for beans of type Queue, Exchange, and Binding and declares them to the broker on behalf of the user. As a result, you need not use that bean explicitly in the simple Java driver. There are plenty of options to configure the properties of the components in the XML schema. You can use auto-complete features of your XML editor to explore them and look at their documentation.

The following example repeats the same example as the preceding example but with the external configuration defined in Java:

Spring Boot automatically configures the infrastructure beans, as the following example shows:

The following classes/interfaces have been moved from org.springframework.amqp.rabbit.core.support to org.springframework.amqp.rabbit.batch:

In addition, ListenerExecutionFailedException has been moved from org.springframework.amqp.rabbit.listener.exception to org.springframework.amqp.rabbit.support.

You can now configure an executor on each listener, overriding the factory configuration, to more easily identify threads associated with the listener. You can now override the container factory’s acknowledgeMode property with the annotation’s ackMode property. See overriding container factory properties for more information.

When using batching, @RabbitListener methods can now receive a complete batch of messages in one call instead of getting them one at at time.

Spring Data Projection interfaces are now supported by the Jackson2JsonMessageConverter. See Using Spring Data Projection Interfaces for more information.

The Log4J and Logback AmqpAppender s now support a verifyHostname SSL option. See Logging Subsystem AMQP Appenders for more information.

The MessageListenerAdapter provides now a new buildListenerArguments(Object, Channel, Message) method to build an array of arguments to be passed into target listener and an old one is deprecated. See MessageListenerAdapter for more information.

The 0-9-1 AMQP specification does not define a Message class or interface. Instead, when performing an operation such as basicPublish(), the content is passed as a byte-array argument and additional properties are passed in as separate arguments. Spring AMQP defines a Message class as part of a more general AMQP domain model representation. The purpose of the Message class is to encapsulate the body and properties within a single instance so that the API can, in turn, be simpler. The following example shows the Message class definition:

The MessageProperties interface defines several common properties, such as 'messageId', 'timestamp', 'contentType', and several more. You can also extend those properties with user-defined 'headers' by calling the setHeader(String key, Object value) method.

The Exchange interface represents an AMQP Exchange, which is what a Message Producer sends to. Each Exchange within a virtual host of a broker has a unique name as well as a few other properties. The following example shows the Exchange interface:

As you can see, an Exchange also has a 'type' represented by constants defined in ExchangeTypes. The basic types are: direct, topic, fanout`, and headers. In the core package, you can find implementations of the Exchange interface for each of those types. The behavior varies across these Exchange types in terms of how they handle bindings to queues. For example, a Direct exchange lets a queue be bound by a fixed routing key (often the queue’s name). A Topic exchange supports bindings with routing patterns that may include the '*' and '#' wildcards for 'exactly-one' and 'zero-or-more', respectively. The Fanout exchange publishes to all queues that are bound to it without taking any routing key into consideration. For much more information about these and the other Exchange types, see Other Resources.

The Queue class represents the component from which a message consumer receives messages. Like the various Exchange classes, our implementation is intended to be an abstract representation of this core AMQP type. The following listing shows the Queue class:

Notice that the constructor takes the queue name. Depending on the implementation, the admin template may provide methods for generating a uniquely named queue. Such queues can be useful as a “reply-to” address or in other temporary situations. For that reason, the 'exclusive' and 'autoDelete' properties of an auto-generated queue would both be set to 'true'.

Given that a producer sends to an exchange and a consumer receives from a queue, the bindings that connect queues to exchanges are critical for connecting those producers and consumers via messaging. In Spring AMQP, we define a Binding class to represent those connections. This section reviews the basic options for binding queues to exchanges.

You can bind a queue to a DirectExchange with a fixed routing key, as the following example shows:

You can bind a queue to a TopicExchange with a routing pattern, as the following example shows:

You can bind a queue to a FanoutExchange with no routing key, as the following example shows:

We also provide a BindingBuilder to facilitate a “fluent API” style, as the following example shows:

By itself, an instance of the Binding class only holds the data about a connection. In other words, it is not an “active” component. However, as you will see later in Configuring the Broker, the AmqpAdmin class can use Binding instances to actually trigger the binding actions on the broker. Also, as you can see in that same section, you can define the Binding instances by using Spring’s @Bean annotations within @Configuration classes. There is also a convenient base class that further simplifies that approach for generating AMQP-related bean definitions and recognizes the queues, exchanges, and bindings so that they are all declared on the AMQP broker upon application startup.

The AmqpTemplate is also defined within the core package. As one of the main components involved in actual AMQP messaging, it is discussed in detail in its own section (see AmqpTemplate).

Starting with version 1.7, a ConnectionNameStrategy is provided for the injection into the AbstractionConnectionFactory. The generated name is used for the application-specific identification of the target RabbitMQ connection. The connection name is displayed in the management UI if the RabbitMQ server supports it. This value does not have to be unique and cannot be used as a connection identifier — for example, in HTTP API requests. This value is supposed to be human-readable and is a part of ClientProperties under the connection_name key. You can use a simple Lambda, as follows:

The ConnectionFactory argument can be used to distinguish target connection names by some logic. By default, the beanName of the AbstractConnectionFactory, a hex string representing the object, and an internal counter are used to generate the connection_name. The <rabbit:connection-factory> namespace component is also supplied with the connection-name-strategy attribute.

An implementation of SimplePropertyValueConnectionNameStrategy sets the connection name to an application property. You can declare it as a @Bean and inject it into the connection factory, as the following example shows:

The property must exist in the application context’s Environment.

The connection might be blocked for interaction from the broker that corresponds to the Memory Alarm. Starting with version 2.0, the org.springframework.amqp.rabbit.connection.Connection can be supplied with com.rabbitmq.client.BlockedListener instances to be notified for connection blocked and unblocked events. In addition, the AbstractConnectionFactory emits a ConnectionBlockedEvent and ConnectionUnblockedEvent, respectively, through its internal BlockedListener implementation. These let you provide application logic to react appropriately to problems on the broker and (for example) take some corrective actions.

Starting with version 2.0.2, the RabbitTemplate has a configuration option to automatically use a second connection factory, unless transactions are being used. See Using a Separate Connection for more information. The ConnectionNameStrategy for the publisher connection is the same as the primary strategy with .publisher appended to the result of calling the method.

Starting with version 1.7.7, an AmqpResourceNotAvailableException is provided, which is thrown when SimpleConnection.createChannel() cannot create a Channel (for example, because the channelMax limit is reached and there are no available channels in the cache). You can use this exception in the RetryPolicy to recover the operation after some back-off.

The CachingConnectionFactory uses an instance of the Rabbit client ConnectionFactory. A number of configuration properties are passed through (host, port, userName, password, requestedHeartBeat, and connectionTimeout for example) when setting the equivalent property on the CachingConnectionFactory. To set other properties (clientProperties, for example), you can define an instance of the Rabbit factory and provide a reference to it by using the appropriate constructor of the CachingConnectionFactory. When using the namespace (as described earlier), you need to provide a reference to the configured factory in the connection-factory attribute. For convenience, a factory bean is provided to assist in configuring the connection factory in a Spring application context, as discussed in the next section.

Starting with version 1.4, a convenient RabbitConnectionFactoryBean is provided to enable convenient configuration of SSL properties on the underlying client connection factory by using dependency injection. Other setters delegate to the underlying factory. Previously, you had to configure the SSL options programmatically. The following example shows how to configure a RabbitConnectionFactoryBean:

See the RabbitMQ Documentation for information about configuring SSL. Omit the keyStore and trustStore configuration to connect over SSL without certificate validation. The next example shows how you can provide key and trust store configuration.

The sslPropertiesLocation property is a Spring Resource pointing to a properties file containing the following keys:

The keyStore and truststore are Spring Resources pointing to the stores. Typically this properties file is secured by the operating system with the application having read access.

Starting with Spring AMQP version 1.5,you can set these properties directly on the factory bean. If both discrete properties and sslPropertiesLocation is provided, properties in the latter override the discrete values.

Starting with version 1.3, the AbstractRoutingConnectionFactory has been introduced. This factory provides a mechanism to configure mappings for several ConnectionFactories and determine a target ConnectionFactory by some lookupKey at runtime. Typically, the implementation checks a thread-bound context. For convenience, Spring AMQP provides the SimpleRoutingConnectionFactory, which gets the current thread-bound lookupKey from the SimpleResourceHolder. The following examples shows how to configure a SimpleRoutingConnectionFactory in both XML and Java:

It is important to unbind the resource after use. For more information, see the JavaDoc for AbstractRoutingConnectionFactory.

Starting with version 1.4, RabbitTemplate supports the SpEL sendConnectionFactorySelectorExpression and receiveConnectionFactorySelectorExpression properties, which are evaluated on each AMQP protocol interaction operation (send, sendAndReceive, receive, or receiveAndReply), resolving to a lookupKey value for the provided AbstractRoutingConnectionFactory. You can use bean references, such as @vHostResolver.getVHost(#root) in the expression. For send operations, the message to be sent is the root evaluation object. For receive operations, the queueName is the root evaluation object.

The routing algorithm is as follows: If the selector expression is null or is evaluated to null or the provided ConnectionFactory is not an instance of AbstractRoutingConnectionFactory, everything works as before, relying on the provided ConnectionFactory implementation. The same occurs if the evaluation result is not null, but there is no target ConnectionFactory for that lookupKey and the AbstractRoutingConnectionFactory is configured with lenientFallback = true. In the case of an AbstractRoutingConnectionFactory, it does fallback to its routing implementation based on determineCurrentLookupKey(). However, if lenientFallback = false, an IllegalStateException is thrown.

The namespace support also provides the send-connection-factory-selector-expression and receive-connection-factory-selector-expression attributes on the <rabbit:template> component.

Also, starting with version 1.4, you can configure a routing connection factory in a listener container. In that case, the list of queue names is used as the lookup key. For example, if you configure the container with setQueueNames("thing1", "thing2"), the lookup key is [thing1,thing]" (note that there is no space in the key).

Starting with version 1.6.9, you can add a qualifier to the lookup key by using setLookupKeyQualifier on the listener container. Doing so enables, for example, listening to queues with the same name but in a different virtual host (where you would have a connection factory for each).

For example, with lookup key qualifier thing1 and a container listening to queue thing2, the lookup key you could register the target connection factory with could be thing1[thing2].

When using HA queues in a cluster, for the best performance, you may want to connect to the physical broker where the master queue resides. While the CachingConnectionFactory can be configured with multiple broker addresses. This is to fail over and the client attempts to connect in order. The LocalizedQueueConnectionFactory uses the REST API provided by the admin plugin to determine on which node the queue is mastered. It then creates (or retrieves from a cache) a CachingConnectionFactory that connects to just that node. If the connection fails, the new master node is determined and the consumer connects to it. The LocalizedQueueConnectionFactory is configured with a default connection factory, in case the physical location of the queue cannot be determined, in which case it connects as normal to the cluster.

The LocalizedQueueConnectionFactory is a RoutingConnectionFactory and the SimpleMessageListenerContainer uses the queue names as the lookup key as discussed in Routing Connection Factory above.

The following example configuration shows how to use Spring Boot’s RabbitProperties to configure the factories:

Notice that the first three parameters are arrays of addresses, adminUris, and nodes. These are positional in that, when a container attempts to connect to a queue, it determines on which node the queue is mastered and connects to the address in the same array position.

Confirmed and returned messages are supported by setting the publisherConfirms and publisherReturns properties of the CachingConnectionFactory to 'true'.

When these options are set, Channel instances created by the factory are wrapped in an PublisherCallbackChannel, which is used to facilitate the callbacks. When such a channel is obtained, the client can register a PublisherCallbackChannel.Listener with the Channel. The PublisherCallbackChannel implementation contains logic to route a confirm or return to the appropriate listener. These features are explained further in the following sections.

See also simplePublisherConfirms in Scoped Operations.

The connection factory supports registering ConnectionListener and ChannelListener implementations. This allows you to receive notifications for connection and channel related events. (A ConnectionListener is used by the RabbitAdmin to perform declarations when the connection is established - see Automatic Declaration of Exchanges, Queues, and Bindings for more information). The following listing shows the ConnectionListener interface definition:

Starting with version 2.0, the org.springframework.amqp.rabbit.connection.Connection object can be supplied with com.rabbitmq.client.BlockedListener instances to be notified for connection blocked and unblocked events. The following example shows the ChannelListener interface definition:

See Publishing is Asynchronous — How to Detect Successes and Failures for one scenario where you might want to register a ChannelListener.

Version 1.5 introduced a mechanism to enable users to control logging levels.

The CachingConnectionFactory uses a default strategy to log channel closures as follows:

To modify this behavior, you can inject a custom ConditionalExceptionLogger into the CachingConnectionFactory in its closeExceptionLogger property.

See also Consumer Events.

Staring with version 1.6, the CachingConnectionFactory now provides cache statistics through the getCacheProperties() method. These statistics can be used to tune the cache to optimize it in production. For example, the high water marks can be used to determine whether the cache size should be increased. If it equals the cache size, you might want to consider increasing further. The following table describes the CacheMode.CHANNEL properties:

The following table describes the CacheMode.CONNECTION properties:

The cacheMode property (CHANNEL or CONNECTION) is also included.

Since the first version of Spring AMQP, the framework has provided its own connection and channel recovery in the event of a broker failure. Also, as discussed in Configuring the Broker, the RabbitAdmin re-declares any infrastructure beans (queues and others) when the connection is re-established. It therefore does not rely on the auto-recovery that is now provided by the amqp-client library. Spring AMQP now uses the 4.0.x version of amqp-client, which has auto recovery enabled by default. Spring AMQP can still use its own recovery mechanisms if you wish, disabling it in the client, (by setting the automaticRecoveryEnabled property on the underlying RabbitMQ connectionFactory to false). However, the framework is completely compatible with auto-recovery being enabled. This means any consumers you create within your code (perhaps via RabbitTemplate.execute()) can be recovered automatically.

Starting with version 1.3, you can now configure the RabbitTemplate to use a RetryTemplate to help with handling problems with broker connectivity. See the spring-retry project for complete information. The following is only one example that uses an exponential back off policy and the default SimpleRetryPolicy, which makes three tries before throwing the exception to the caller.

The following example uses the XML namespace:

The following example uses the @Configuration annotation in Java:

Starting with version 1.4, in addition to the retryTemplate property, the recoveryCallback option is supported on the RabbitTemplate. It is used as a second argument for the RetryTemplate.execute(RetryCallback<T, E> retryCallback, RecoveryCallback<T> recoveryCallback).

In this case, you would not inject a RetryTemplate into the RabbitTemplate.

Publishing messages is an asynchronous mechanism and, by default, messages that cannot be routed are dropped by RabbitMQ. For successful publishing, you can receive an asynchronous confirm, as described in Publisher Confirms and Returns. Consider two failure scenarios:

The first case is covered by publisher returns, as described in Publisher Confirms and Returns.

For the second case, the message is dropped and no return is generated. The underlying channel is closed with an exception. By default, this exception is logged, but you can register a ChannelListener with the CachingConnectionFactory to obtain notifications of such events. The following example shows how to add a ConnectionListener:

You can examine the signal’s reason property to determine the problem that occurred.

To detect the exception on the sending thread, you can setChannelTransacted(true) on the RabbitTemplate and the exception is detected on the txCommit(). However, transactions significantly impede performance, so consider this carefully before enabling transactions for just this one use case.

The RabbitTemplate implementation of AmqpTemplate supports publisher confirms and returns.

For returned messages, the template’s mandatory property must be set to true or the mandatory-expression must evaluate to true for a particular message. This feature requires a CachingConnectionFactory that has its publisherReturns property set to true (see Publisher Confirms and Returns). Returns are sent to the client by it registering a RabbitTemplate.ReturnCallback by calling setReturnCallback(ReturnCallback callback). The callback must implement the following method:

Only one ReturnCallback is supported by each RabbitTemplate. See also Reply Timeout.

For publisher confirms (also known as publisher acknowledgements), the template requires a CachingConnectionFactory that has its publisherConfirms property set to true. Confirms are sent to the client by it registering a RabbitTemplate.ConfirmCallback by calling setConfirmCallback(ConfirmCallback callback). The callback must implement this method:

The CorrelationData is an object supplied by the client when sending the original message. The ack is true for an ack and false for a nack. For nack instances, the cause may contain a reason for the nack, if it is available when the nack is generated. An example is when sending a message to a non-existent exchange. In that case, the broker closes the channel. The reason for the closure is included in the cause. The cause was added in version 1.4.

Only one ConfirmCallback is supported by a RabbitTemplate.

Starting with version 2.1, the CorrelationData object has a ListenableFuture that you can \used to get the result, instead of using a ConfirmCallback on the template. The following example shows how to configure a CorrelationData instance:

Since it is a ListenableFuture<Confirm>, you can either get() the result when ready or add listeners for an asynchronous callback. The Confirm object is a simple bean with 2 properties: ack and reason (for nack instances). The reason is not populated for broker-generated nack instances. It is populated for nack instances generated by the framework (for example, closing the connection while ack instances are outstanding).

In addition, when both confirms and returns are enabled, the CorrelationData is populated with the returned message. It is guaranteed that this occurs before the future is set with the ack.

See also Scoped Operations for a simpler mechanism for waiting for publisher confirms.

Normally, when using the template, a Channel is checked out of the cache (or created), used for the operation, and returned to the cache for reuse. In a multi-threaded environment, there is no guarantee that the next operation uses the same channel. There may be times, however, where you want to have more control over the use of a channel and ensure that a number of operations are all performed on the same channel.

Starting with version 2.0, a new method called invoke is provided, with an OperationsCallback. Any operations performed within the scope of the callback and on the provided RabbitOperations argument use the same dedicated Channel, which will be closed at the end (not returned to a cache). If the channel is a PublisherCallbackChannel, it is returned to the cache after all confirms have been received (see Publisher Confirms and Returns).

One example of why you might need this is if you wish to use the waitForConfirms() method on the underlying Channel. This method was not previously exposed by the Spring API because the channel is, generally, cached and shared, as discussed earlier. The RabbitTemplate now provides waitForConfirms(long timeout) and waitForConfirmsOrDie(long timeout), which delegate to the dedicated channel used within the scope of the OperationsCallback. The methods cannot be used outside of that scope, for obvious reasons.

Note that a higher-level abstraction that lets you correlate confirms to requests is provided elsewhere (see Publisher Confirms and Returns). If you want only to wait until the broker has confirmed delivery, you can use the technique shown in the following example:

If you wish RabbitAdmin operations to be invoked on the same channel within the scope of the OperationsCallback, the admin must have been constructed by using the same RabbitTemplate that was used for the invoke operation.

When using confirms in this way, much of the infrastructure set up for correlating confirms to requests is not really needed. Starting with version 2.1, the connection factory supports a new property called simplePublisherConfirms. When this is true, the infrastructure is avoided and the confirm processing can be more efficient.

Furthermore, the RabbitTemplate sets the publisherSequenceNumber property in the sent message MessageProperties. If you wish to check (or log or otherwise use) specific confirms, you can do so with an overloaded invoke method, as the following example shows:

The following example logs ack and nack instances:

Starting with version 1.4, RabbitMessagingTemplate (built on top of RabbitTemplate) provides an integration with the Spring Framework messaging abstraction — that is, org.springframework.messaging.Message. This lets you send and receive messages by using the spring-messaging Message<?> abstraction. This abstraction is used by other Spring projects, such as Spring Integration and Spring’s STOMP support. There are two message converters involved: one to convert between a spring-messaging Message<?> and Spring AMQP’s Message abstraction and one to convert between Spring AMQP’s Message abstraction and the format required by the underlying RabbitMQ client library. By default, the message payload is converted by the provided RabbitTemplate instance’s message converter. Alternatively, you can inject a custom MessagingMessageConverter with some other payload converter, as the following example shows:

Starting with version 1.6, the template now supports a user-id-expression (userIdExpression when using Java configuration). If a message is sent, the user id property is set (if not already set) after evaluating this expression. The root object for the evaluation is the message to be sent.

The following examples show how to use the user-id-expression attribute:

The first example is a literal expression. The second obtains the username property from a connection factory bean in the application context.

Starting with version 2.0.2, you can set the usePublisherConnection property to true to use a different connection to that used by listener containers, when possible. This is to avoid consumers being blocked when a producer is blocked for any reason. The CachingConnectionFactory now maintains a second internal connection factory for this purpose. If the rabbit template is running in a transaction started by the listener container, the container’s channel is used, regardless of this setting.

Starting with version 1.3, a message builder API is provided by the MessageBuilder and MessagePropertiesBuilder. These methods provide a convenient “fluent” means of creating a message or message properties. The following examples show the fluent API in action:

Each of the properties defined on the MessageProperties can be set. Other methods include setHeader(String key, String value), removeHeader(String key), removeHeaders(), and copyProperties(MessageProperties properties). Each property setting method has a set*IfAbsent() variant. In the cases where a default initial value exists, the method is named set*IfAbsentOrDefault().

Five static methods are provided to create an initial message builder:

Three static methods are provided to create a MessagePropertiesBuilder instance:

With the RabbitTemplate implementation of AmqpTemplate, each of the send() methods has an overloaded version that takes an additional CorrelationData object. When publisher confirms are enabled, this object is returned in the callback described in AmqpTemplate. This lets the sender correlate a confirm (ack or nack) with the sent message.

Starting with version 1.6.7, the CorrelationAwareMessagePostProcessor interface was introduced, allowing the correlation data to be modified after the message has been converted. The following example shows how to use it:

In version 2.0, this interface is deprecated. The method has been moved to MessagePostProcessor with a default implementation that delegates to postProcessMessage(Message message).

Also starting with version 1.6.7, a new callback interface called CorrelationDataPostProcessor is provided. This is invoked after all MessagePostProcessor instances (provided in the send() method as well as those provided in setBeforePublishPostProcessors()). Implementations can update or replace the correlation data supplied in the send() method (if any). The Message and original CorrelationData (if any) are provided as arguments. The following example shows how to use the postProcess method:

When the template’s mandatory property is true, returned messages are provided by the callback described in AmqpTemplate.

Starting with version 1.4, the RabbitTemplate supports the SpEL mandatoryExpression property, which is evaluated against each request message as the root evaluation object, resolving to a boolean value. Bean references, such as @myBean.isMandatory(#root), can be used in the expression.

Publisher returns can also be used internally by the RabbitTemplate in send and receive operations. See Reply Timeout for more information.

Version 1.4.2 introduced the BatchingRabbitTemplate. This is a subclass of RabbitTemplate with an overridden send method that batches messages according to the BatchingStrategy. Only when a batch is complete is the message sent to RabbitMQ. The following listing shows the BatchingStrategy interface definition:

A SimpleBatchingStrategy is provided. It supports sending messages to a single exchange or routing key. It has the following properties:

The SimpleBatchingStrategy formats the batch by preceding each embedded message with a four-byte binary length. This is communicated to the receiving system by setting the springBatchFormat message property to lengthHeader4.

However, see @RabbitListener with Batching for more information.

The AmqpTemplate itself can be used for polled Message reception. By default, if no message is available, null is returned immediately. There is no blocking. Starting with version 1.5, you can set a receiveTimeout, in milliseconds, and the receive methods block for up to that long, waiting for a message. A value less than zero means block indefinitely (or at least until the connection to the broker is lost). Version 1.6 introduced variants of the receive methods that let the timeout be passed in on each call.

There are four simple receive methods available. As with the Exchange on the sending side, there is a method that requires that a default queue property has been set directly on the template itself, and there is a method that accepts a queue parameter at runtime. Version 1.6 introduced variants to accept timeoutMillis to override receiveTimeout on a per-request basis. The following listing shows the definitions of the four methods:

As in the case of sending messages, the AmqpTemplate has some convenience methods for receiving POJOs instead of Message instances, and implementations provide a way to customize the MessageConverter used to create the Object returned: The following listing shows those methods:

Starting with version 2.0, there are variants of these methods that take an additional ParameterizedTypeReference argument to convert complex types. The template must be configured with a SmartMessageConverter. See Converting From a Message With RabbitTemplate for more information.

Similar to sendAndReceive methods, beginning with version 1.3, the AmqpTemplate has several convenience receiveAndReply methods for synchronously receiving, processing and replying to messages. The following listing shows those method definitions:

The AmqpTemplate implementation takes care of the receive and reply phases. In most cases, you should provide only an implementation of ReceiveAndReplyCallback to perform some business logic for the received message and build a reply object or message, if needed. Note, a ReceiveAndReplyCallback may return null. In this case, no reply is sent and receiveAndReply works like the receive method. This lets the same queue be used for a mixture of messages, some of which may not need a reply.

Automatic message (request and reply) conversion is applied only if the provided callback is not an instance of ReceiveAndReplyMessageCallback, which provides a raw message exchange contract.

The ReplyToAddressCallback is useful for cases requiring custom logic to determine the replyTo address at runtime against the received message and reply from the ReceiveAndReplyCallback. By default, replyTo information in the request message is used to route the reply.

The following listing shows an example of POJO-based receive and reply:

Batched messages are automatically de-batched by listener containers (using the springBatchFormat message header). Rejecting any message from a batch causes the entire batch to be rejected. See Batching for more information about batching.

The containers publish application events whenever a listener (consumer) experiences a failure of some kind. The event ListenerContainerConsumerFailedEvent has the following properties:

These events can be consumed by implementing ApplicationListener<ListenerContainerConsumerFailedEvent>.

If a consumer fails because one if its queues is being used exclusively, by default, as well as publishing the event, a WARN log is issued. To change this logging behavior, provide a custom ConditionalExceptionLogger in the SimpleMessageListenerContainer instance’s exclusiveConsumerExceptionLogger property. See also Logging Channel Close Events.

Fatal errors are always logged at the ERROR level. This it not modifiable.

Several other events are published at various stages of the container lifecycle:

You can provide a strategy to generate consumer tags. By default, the consumer tag is generated by the broker. The following listing shows the ConsumerTagStrategy interface definition:

The queue is made available so that it can (optionally) be used in the tag.

See Message Listener Container Configuration.

The easiest way to receive a message asynchronously is to use the annotated listener endpoint infrastructure. In a nutshell, it lets you expose a method of a managed bean as a Rabbit listener endpoint. The following example shows how to use the @RabbitListener annotation:

The idea of the preceding example is that, whenever a message is available on the queue named myQueue, the processOrder method is invoked accordingly (in this case, with the payload of the message).

The annotated endpoint infrastructure creates a message listener container behind the scenes for each annotated method, by using a RabbitListenerContainerFactory.

In the preceding example, myQueue must already exist and be bound to some exchange. The queue can be declared and bound automatically, as long as a RabbitAdmin exists in the application context.

In the first example, a queue myQueue is declared automatically (durable) together with the exchange, if needed, and bound to the exchange with the routing key. In the second example, an anonymous (exclusive, auto-delete) queue is declared and bound. Multiple QueueBinding entries can be provided, letting the listener listen to multiple queues. In the third example, a queue with the name retrieved from property my.queue is declared, if necessary, with the default binding to the default exchange using the queue name as the routing key.

Since version 2.0, the @Exchange annotation supports any exchange types, including custom. For more information, see AMQP Concepts.

You can use normal @Bean definitions when you need more advanced configuration.

Notice ignoreDeclarationExceptions on the exchange in the first example. This allows, for example, binding to an existing exchange that might have different settings (such as internal). By default, the properties of an existing exchange must match.

Starting with version 2.0, you can now bind a queue to an exchange with multiple routing keys, as the following example shows:

You can also specify arguments within @QueueBinding annotations for queues, exchanges, and bindings, as the following example shows:

Notice that the x-message-ttl argument is set to 10 seconds for the queue. Since the argument type is not String, we have to specify its type — in this case, Integer. As with all such declarations, if the queue already exists, the arguments must match those on the queue. For the header exchange, we set the binding arguments to match messages that have the thing1 header set to somevalue, and the thing2 header must be present with any value. The x-match argument means both conditions must be satisfied.

The argument name, value, and type can be property placeholders (${…​}) or SpEL expressions (#{…​}). The name must resolve to a String. The expression for type must resolve to a Class or the fully-qualified name of a class. The value must resolve to something that can be converted by the DefaultConversionService to the type (such as the x-message-ttl in the preceding example).

If a name resolves to null or an empty String, that @Argument is ignored.

When receiving a a batch of messages, the de-batching is normally performed by the container and the listener is invoked with one message at at time. Starting with version 2.2, you can configure the listener container factory and listener to receive the entire batch in one call, simply set the factory’s batchListener property, and make the method payload parameter a List:

Setting the batchListener property to true automatically turns off the debatchingEnabled container property in containers that the factory creates - effectively, the debatching is moved from the container to the listener adapter and the adapter creates the list that is passed to the listener.

A batch-enabled factory cannot be used with a multi-method listener.

Listener container factories were introduced to support the @RabbitListener and registering containers with the RabbitListenerEndpointRegistry, as discussed in Programmatic Endpoint Registration.

Starting with version 2.1, they can be used to create any listener container — even a container without a listener (such as for use in Spring Integration). Of course, a listener must be added before the container is started.

There are two ways to create such containers:

The following example shows how to use a SimpleRabbitListenerEndpoint to create a listener container:

The following example shows how to add the listener after creation:

In either case, the listener can also be a ChannelAwareMessageListener, since it is now a sub-interface of MessageListener.

These techniques are useful if you wish to create several containers with similar properties or use a pre-configured container factory such as the one provided by Spring Boot auto configuration or both.

Starting with version 2.1, @RabbitListener (and @RabbitHandler) methods can be specified with asynchronous return types ListenableFuture<?> and Mono<?>, letting the reply be sent asynchronously.

A number of different threads are involved with asynchronous consumers.

Threads from the TaskExecutor configured in the SimpleMessageListenerContainer are used to invoke the MessageListener when a new message is delivered by RabbitMQ Client. If not configured, a SimpleAsyncTaskExecutor is used. If you use a pooled executor, you need to ensure the pool size is sufficient to handle the configured concurrency. With the DirectMessageListenerContainer, the MessageListener is invoked directly on a RabbitMQ Client thread. In this case, the taskExecutor is used for the task that monitors the consumers.

The Executor configured in the CachingConnectionFactory is passed into the RabbitMQ Client when creating the connection, and its threads are used to deliver new messages to the listener container. If this is not configured, the client uses an internal thread pool executor with a pool size of five.

The RabbitMQ client uses a ThreadFactory to create threads for low-level I/O (socket) operations. To modify this factory, you need to configure the underlying RabbitMQ ConnectionFactory, as discussed in Configuring the Underlying Client Connection Factory.

Version 2.0 introduced the DirectMessageListenerContainer (DMLC). Previously, only the SimpleMessageListenerContainer (SMLC) was available. The SMLC uses an internal queue and a dedicated thread for each consumer. If a container is configured to listen to multiple queues, the same consumer thread is used to process all the queues. Concurrency is controlled by concurrentConsumers and other properties. As messages arrive from the RabbitMQ client, the client thread hands them off to the consumer thread through the queue. This architecture was required because, in early versions of the RabbitMQ client, multiple concurrent deliveries were not possible. Newer versions of the client have a revised threading model and can now support concurrency. This has allowed the introduction of the DMLC where the listener is now invoked directly on the RabbitMQ Client thread. Its architecture is, therefore, actually “simpler” than the SMLC. However, there are some limitations with this approach, and certain features of the SMLC are not available with the DMLC. Also, concurrency is controlled by consumersPerQueue (and the client library’s thread pool). The concurrentConsumers and associated properties are not available with this container.

The following features are available with the SMLC but not the DMLC:

However, the DMLC has the following benefits over the SMLC:

See Message Listener Container Configuration for information about which configuration properties apply to each container.

While efficient, one problem with asynchronous consumers is detecting when they are idle — users might want to take some action if no messages arrive for some period of time.

Starting with version 1.6, it is now possible to configure the listener container to publish a ListenerContainerIdleEvent when some time passes with no message delivery. While the container is idle, an event is published every idleEventInterval milliseconds.

To configure this feature, set idleEventInterval on the container. The following example shows how to do so in XML and in Java (for both a SimpleMessageListenerContainer and a SimpleRabbitListenerContainerFactory):

In each of these cases, an event is published once per minute while the container is idle.

The default implementation of the MessageConverter strategy is called SimpleMessageConverter. This is the converter that is used by an instance of RabbitTemplate if you do not explicitly configure an alternative. It handles text-based content, serialized Java objects, and byte arrays.

This converter is similar to the SimpleMessageConverter except that it can be configured with other Spring Framework Serializer and Deserializer implementations for application/x-java-serialized-object conversions.

See Java Deserialization for important information.

This section covers using the Jackson2JsonMessageConverter to convert to and from a Message. It has the following sections:

Yet another option is the MarshallingMessageConverter. It delegates to the Spring OXM library’s implementations of the Marshaller and Unmarshaller strategy interfaces. You can read more about that library here. In terms of configuration, it is most common to provide only the constructor argument, since most implementations of Marshaller also implement Unmarshaller. The following example shows how to configure a MarshallingMessageConverter:

This class was introduced in version 2.1 and can be used to convert messages from and to XML.

Both Jackson2XmlMessageConverter and Jackson2JsonMessageConverter have the same base class: AbstractJackson2MessageConverter.

The Jackson2XmlMessageConverter uses the com.fasterxml.jackson 2.x library.

You can use it the same way as Jackson2JsonMessageConverter, except it supports XML instead of JSON. The following example configures a Jackson2JsonMessageConverter:

See Jackson2JsonMessageConverter for more information.

This class was introduced in version 1.4.2 and allows delegation to a specific MessageConverter based on the content type property in the MessageProperties. By default, it delegates to a SimpleMessageConverter if there is no contentType property or there is a value that matches none of the configured converters. The following example configures a ContentTypeDelegatingMessageConverter:

This section covers how to deserialize Java objects.

The MessagePropertiesConverter strategy interface is used to convert between the Rabbit Client BasicProperties and Spring AMQP MessageProperties. The default implementation (DefaultMessagePropertiesConverter) is usually sufficient for most purposes, but you can implement your own if needed. The default properties converter converts BasicProperties elements of type LongString to String instances when the size is not greater than 1024 bytes. Larger LongString instances are not converted (see the next paragraph). This limit can be overridden with a constructor argument.

Starting with version 1.6, headers longer than the long string limit (default: 1024) are now left as LongString instances by default by the DefaultMessagePropertiesConverter. You can access the contents through the getBytes[], toString(), or getStream() methods.

Previously, the DefaultMessagePropertiesConverter “converted” such headers to a DataInputStream (actually it just referenced the LongString instance’s DataInputStream). On output, this header was not converted (except to a String — for example, java.io.DataInputStream@1d057a39 by calling toString() on the stream).

Large incoming LongString headers are now correctly “converted” on output, too (by default).

A new constructor is provided to let you configure the converter to work as before. The following listing shows the Javadoc comment and declaration of the method:

Also starting with version 1.6, a new property called correlationIdString has been added to MessageProperties. Previously, when converting to and from BasicProperties used by the RabbitMQ client, an unnecessary byte[] <→ String conversion was performed because MessageProperties.correlationId is a byte[], but BasicProperties uses a String. (Ultimately, the RabbitMQ client uses UTF-8 to convert the String to bytes to put in the protocol message).

To provide maximum backwards compatibility, a new property called correlationIdPolicy has been added to the DefaultMessagePropertiesConverter. This takes a DefaultMessagePropertiesConverter.CorrelationIdPolicy enum argument. By default it is set to BYTES, which replicates the previous behavior.

For inbound messages:

For outbound messages:

Also starting with version 1.6, the inbound deliveryMode property is no longer mapped to MessageProperties.deliveryMode. It is mapped to MessageProperties.receivedDeliveryMode instead. Also, the inbound userId property is no longer mapped to MessageProperties.userId. It is mapped to MessageProperties.receivedUserId instead. These changes are to avoid unexpected propagation of these properties if the same MessageProperties object is used for an outbound message.

By default, the send and receive methods timeout after five seconds and return null. You can modify this behavior by setting the replyTimeout property. Starting with version 1.5, if you set the mandatory property to true (or the mandatory-expression evaluates to true for a particular message), if the message cannot be delivered to a queue, an AmqpMessageReturnedException is thrown. This exception has returnedMessage, replyCode, and replyText properties, as well as the exchange and routingKey used for the send.

Starting with version 2.1.2, a replyTimedOut method has been added, letting subclasses be informed of the timeout so that they can clean up any retained state.

Starting with versions 2.0.11 and 2.1.3, when you use the default DirectReplyToMessageListenerContainer, you can add an error handler by setting the template’s replyErrorHandler property. This error handler is invoked for any failed deliveries, such as late replies and messages received without a correlation header. The exception passed in is a ListenerExecutionFailedException, which has a failedMessage property.

Reply listeners are still supported with named queues (other than amq.rabbitmq.reply-to), allowing control of reply concurrency and so on.

Starting with version 1.6, if you wish to use a temporary, exclusive, auto-delete queue for each reply, set the useTemporaryReplyQueues property to true. This property is ignored if you set a replyAddress.

You can change the criteria that dictate whether to use direct reply-to by subclassing RabbitTemplate and overriding useDirectReplyTo() to check different criteria. The method is called once only, when the first request is sent.

Prior to version 2.0, the RabbitTemplate created a new consumer for each request and canceled the consumer when the reply was received (or timed out). Now the template uses a DirectReplyToMessageListenerContainer instead, letting the consumers be reused. The template still takes care of correlating the replies, so there is no danger of a late reply going to a different sender. If you want to revert to the previous behavior, set the useDirectReplyToContainer (direct-reply-to-container when using XML configuration) property to false.

The AsyncRabbitTemplate has no such option. It always used a DirectReplyToContainer for replies when direct reply-to is used.

When using a fixed reply queue (other than amq.rabbitmq.reply-to), you must provide correlation data so that replies can be correlated to requests. See RabbitMQ Remote Procedure Call (RPC). By default, the standard correlationId property is used to hold the correlation data. However, if you wish to use a custom property to hold correlation data, you can set the correlation-key attribute on the <rabbit-template/>. Explicitly setting the attribute to correlationId is the same as omitting the attribute. The client and server must use the same header for correlation data.

By default, the template generates its own correlation ID (ignoring any user-supplied value). If you wish to use your own correlation ID, set the RabbitTemplate instance’s userCorrelationId property to true.

When using RabbitMQ versions prior to 3.4.0, a new temporary queue is used for each reply. However, a single reply queue can be configured on the template, which can be more efficient and also lets you set arguments on that queue. In this case, however, you must also provide a <reply-listener/> sub element. This element provides a listener container for the reply queue, with the template being the listener. All of the Message Listener Container Configuration attributes allowed on a <listener-container/> are allowed on the element, except for connection-factory and message-converter, which are inherited from the template’s configuration.

The following example defines a rabbit template with a connection factory:

While the container and template share a connection factory, they do not share a channel. Therefore, requests and replies are not performed within the same transaction (if transactional).

With this configuration, a SimpleListenerContainer is used to receive the replies, with the RabbitTemplate being the MessageListener. When defining a template with the <rabbit:template/> namespace element, as shown in the preceding example, the parser defines the container and wires in the template as the listener.

If you define your RabbitTemplate as a <bean/> or use an @Configuration class to define it as an @Bean or when you create the template programmatically, you need to define and wire up the reply listener container yourself. If you fail to do this, the template never receives the replies and eventually times out and returns null as the reply to a call to a sendAndReceive method.

Starting with version 1.5, the RabbitTemplate detects if it has been configured as a MessageListener to receive replies. If not, attempts to send and receive messages with a reply address fail with an IllegalStateException (because the replies are never received).

Further, if a simple replyAddress (queue name) is used, the reply listener container verifies that it is listening to a queue with the same name. This check cannot be performed if the reply address is an exchange and routing key and a debug log message is written.

The following listing shows examples of how to manually wire up the beans:

A complete example of a RabbitTemplate wired with a fixed reply queue, together with a “remote” listener container that handles the request and returns the reply is shown in this test case.

Prior to version 1.3.6, late replies for timed out messages were only logged. Now, if a late reply is received, it is rejected (the template throws an AmqpRejectAndDontRequeueException). If the reply queue is configured to send rejected messages to a dead letter exchange, the reply can be retrieved for later analysis. To do so, bind a queue to the configured dead letter exchange with a routing key equal to the reply queue’s name.

See the RabbitMQ Dead Letter Documentation for more information about configuring dead lettering. You can also take a look at the FixedReplyQueueDeadLetterTests test case for an example.

Version 1.6 introduced the AsyncRabbitTemplate. This has similar sendAndReceive (and convertSendAndReceive) methods to those on the AmqpTemplate. However, instead of blocking, they return a ListenableFuture.

The sendAndReceive methods return a RabbitMessageFuture. The convertSendAndReceive methods return a RabbitConverterFuture.

You can either synchronously retrieve the result later, by invoking get() on the future, or you can register a callback that is called asynchronously with the result. The following listing shows both approaches:

If mandatory is set and the message cannot be delivered, the future throws an ExecutionException with a cause of AmqpMessageReturnedException, which encapsulates the returned message and information about the return.

If enableConfirms is set, the future has a property called confirm, which is itself a ListenableFuture<Boolean> with true indicating a successful publish. If the confirm future is false, the RabbitFuture has a further property called nackCause, which contains the reason for the failure, if available.

You can set the receiveTimeout property on the template to time out replies (it defaults to 30000 - 30 seconds). If a timeout occurs, the future is completed with an AmqpReplyTimeoutException.

The template implements SmartLifecycle. Stopping the template while there are pending replies causes the pending Future instances to be canceled.

Starting with version 2.0, the asynchronous template now supports direct reply-to instead of a configured reply queue. To enable this feature, use one of the following constructors:

See RabbitMQ Direct reply-to to use direct reply-to with the synchronous RabbitTemplate.

Version 2.0 introduced variants of these methods (convertSendAndReceiveAsType) that take an additional ParameterizedTypeReference argument to convert complex returned types. You must configure the underlying RabbitTemplate with a SmartMessageConverter. See Converting From a Message With RabbitTemplate for more information.

The Spring Framework has a general remoting capability, allowing Remote Procedure Calls (RPC) that use various transports. Spring-AMQP supports a similar mechanism with a AmqpProxyFactoryBean on the client and a AmqpInvokerServiceExporter on the server. This provides RPC over AMQP. On the client side, a RabbitTemplate is used as described earlier. On the server side, the invoker (configured as a MessageListener) receives the message, invokes the configured service, and returns the reply by using the inbound message’s replyTo information.

You can inject the client factory bean into any bean (by using its serviceInterface). The client can then invoke methods on the proxy, resulting in remote execution over AMQP.

On the server side, the AmqpInvokerServiceExporter has both AmqpTemplate and MessageConverter properties. Currently, the template’s MessageConverter is not used. If you need to supply a custom message converter, you should provide it by setting the messageConverter property. On the client side, you can add a custom message converter to the AmqpTemplate, which is provided to the AmqpProxyFactoryBean by using its amqpTemplate property.

The following listing shows sample client and server configurations:

Starting with version 1.3, you can configure the HeadersExchange to match on multiple headers. You can also specify whether any or all headers must match. The following example shows how to do so:

Starting with version 1.6, you can configure Exchanges with an internal flag (defaults to false) and such an Exchange is properly configured on the Broker through a RabbitAdmin (if one is present in the application context). If the internal flag is true for an exchange, RabbitMQ does not let clients use the exchange. This is useful for a dead letter exchange or exchange-to-exchange binding, where you do not wish the exchange to be used directly by publishers.

To see how to use Java to configure the AMQP infrastructure, look at the Stock sample application, where there is the @Configuration class AbstractStockRabbitConfiguration, which ,in turn has RabbitClientConfiguration and RabbitServerConfiguration subclasses. The following listing shows the code for AbstractStockRabbitConfiguration:

In the Stock application, the server is configured by using the following @Configuration class:

This is the end of the whole inheritance chain of @Configuration classes. The end result is that TopicExchange and Queue are declared to the broker upon application startup. There is no binding of TopicExchange to a queue in the server configuration, as that is done in the client application. The stock request queue, however, is automatically bound to the AMQP default exchange. This behavior is defined by the specification.

The client @Configuration class is a little more interesting. Its declaration follows:

The client declares another queue through the declareQueue() method on the AmqpAdmin. It binds that queue to the market data exchange with a routing pattern that is externalized in a properties file.

Version 1.6 introduces a convenient fluent API for configuring Queue and Exchange objects when using Java configuration. The following example shows how to use it:

See the Javadoc for org.springframework.amqp.core.QueueBuilder and org.springframework.amqp.core.ExchangeBuilder for more information.

Starting with version 2.0, the ExchangeBuilder now creates durable exchanges by default, to be consistent with the simple constructors on the individual AbstractExchange classes. To make a non-durable exchange with the builder, use .durable(false) before invoking .build(). The durable() method with no parameter is no longer provided.

You can wrap collections of Declarable objects (Queue, Exchange, and Binding) in Declarables objects. The RabbitAdmin detects such beans (as well as discrete Declarable beans) in the application context, and declares the contained objects on the broker whenever a connection is established (initially and after a connection failure). The following example shows how to do so:

By default, all queues, exchanges, and bindings are declared by all RabbitAdmin instances (assuming they have auto-startup="true") in the application context.

The classes representing these elements implement Declarable, which has two methods: shouldDeclare() and getDeclaringAdmins(). The RabbitAdmin uses these methods to determine whether a particular instance should actually process the declarations on its Connection.

The properties are available as attributes in the namespace, as shown in the following examples:

Similarly, you can use Java-based @Configuration to achieve the same effect. In the following example, the components are declared by admin1 but not by`admin2`:

The name attribute on <rabbit:queue/> and <rabbit:exchange/> elements reflects the name of the entity in the broker. For queues, if the name is omitted, an anonymous queue is created (see AnonymousQueue).

In versions prior to 2.0, the name was also registered as a bean name alias (similar to name on <bean/> elements).

This caused two problems:

Starting with version 2.0, if you declare one of these elements with both an id and a name attribute, the name is no longer declared as a bean name alias. If you wish to declare a queue and exchange with the same name, you must provide an id.

There is no change if the element has only a name attribute. The bean can still be referenced by the name — for example, in binding declarations. However, you still cannot reference it if the name contains SpEL — you must provide an id for reference purposes.

In general, when you need a uniquely-named, exclusive, auto-delete queue, we recommend that you use the AnonymousQueue instead of broker-defined queue names (using "" as a Queue name causes the broker to generate the queue name).

This is because:

You can control the format of the queue name used by AnonymousQueue instances.

By default, the queue name is prefixed by spring.gen- followed by a base64 representation of the UUID — for example: spring.gen-MRBv9sqISkuCiPfOYfpo4g.

You can provide an AnonymousQueue.NamingStrategy implementation in a constructor argument. The following example shows how to do so:

The first bean generates a queue name prefixed by spring.gen- followed by a base64 representation of the UUID — for example: spring.gen-MRBv9sqISkuCiPfOYfpo4g. The second bean generates a queue name prefixed by something- followed by a base64 representation of the UUID. The third bean generates a name by using only the UUID (no base64 conversion) — for example, f20c818a-006b-4416-bf91-643590fedb0e.

The base64 encoding uses the “URL and Filename Safe Alphabet” from RFC 4648. Trailing padding characters (=) are removed.

You can provide your own naming strategy, whereby you can include other information (such as the application name or client host) in the queue name.

You can specify the naming strategy when you use XML configuration. The naming-strategy attribute is present on the <rabbit:queue> element for a bean reference that implements AnonymousQueue.NamingStrategy. The following examples show how to specify the naming strategy in various ways:

The first example creates names such as spring.gen-MRBv9sqISkuCiPfOYfpo4g. The second example creates names with a String representation of a UUID. The third example creates names such as custom.gen-MRBv9sqISkuCiPfOYfpo4g.

You can also provide your own naming strategy bean.

Starting with version 2.1, anonymous queues are declared with argument x-queue-master-locator set to client-local by default. This ensures that the queue is declared on the node to which the application is connected. You can revert to the previous behavior by calling queue.setMasterLocator(null) after constructing the instance.

Prior to version 1.6.6, adding a rollback rule to a container’s transactionAttribute when using an external transaction manager (such as JDBC) had no effect. Exceptions always rolled back the transaction.

Also, when using a transaction advice in the container’s advice chain, conditional rollback was not very usefulm, because all listener exceptions are wrapped in a ListenerExecutionFailedException.

The first problem has been corrected, and the rules are now applied properly. Further, the ListenerFailedRuleBasedTransactionAttribute is now provided. It is a subclass of RuleBasedTransactionAttribute, with the only difference being that it is aware of the ListenerExecutionFailedException and uses the cause of such exceptions for the rule. This transaction attribute can be used directly in the container or through a transaction advice.

The following example uses this rule:

AMQP transactions apply only to messages and acks sent to the broker. Consequently, when there is a rollback of a Spring transaction and a message has been received, Spring AMQP has to not only rollback the transaction but also manually reject the message (sort of a nack, but that is not what the specification calls it). The action taken on message rejection is independent of transactions and depends on the defaultRequeueRejected property (default: true). For more information about rejecting failed messages, see Message Listeners and the Asynchronous Case.

For more information about RabbitMQ transactions and their limitations, see RabbitMQ Broker Semantics.

The RabbitTransactionManager is an alternative to executing Rabbit operations within, and synchronized with, external transactions. This transaction manager is an implementation of the PlatformTransactionManager interface and should be used with a single Rabbit ConnectionFactory.

Application code is required to retrieve the transactional Rabbit resources through ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactory, boolean) instead of a standard Connection.createChannel() call with subsequent channel creation. When using Spring AMQP’s RabbitTemplate, it will autodetect a thread-bound Channel and automatically participate in its transaction.

With Java Configuration, you can setup a new RabbitTransactionManager by using the following bean:

If you prefer XML configuration, you can declare the following bean in your XML Application Context file:

By default, the listener container starts a single consumer that receives messages from the queues.

When examining the table in the previous section, you can see a number of properties and attributes that control concurrency. The simplest is concurrentConsumers, which creates that (fixed) number of consumers that concurrently process messages.

Prior to version 1.3.0, this was the only setting available and the container had to be stopped and started again to change the setting.

Since version 1.3.0, you can now dynamically adjust the concurrentConsumers property. If it is changed while the container is running, consumers are added or removed as necessary to adjust to the new setting.

In addition, a new property called maxConcurrentConsumers has been added and the container dynamically adjusts the concurrency based on workload. This works in conjunction with four additional properties: consecutiveActiveTrigger, startConsumerMinInterval, consecutiveIdleTrigger, and stopConsumerMinInterval. With the default settings, the algorithm to increase consumers works as follows:

If the maxConcurrentConsumers has not been reached and an existing consumer is active for ten consecutive cycles AND at least 10 seconds has elapsed since the last consumer was started, a new consumer is started. A consumer is considered active if it received at least one message in txSize * receiveTimeout milliseconds.

With the default settings, the algorithm to decrease consumers works as follows:

If there are more than concurrentConsumers running and a consumer detects ten consecutive timeouts (idle) AND the last consumer was stopped at least 60 seconds ago, a consumer is stopped. The timeout depends on the receiveTimeout and the txSize properties. A consumer is considered idle if it receives no messages in txSize * receiveTimeout milliseconds. So, with the default timeout (one second) and a txSize of four, stopping a consumer is considered after 40 seconds of idle time (four timeouts correspond to one idle detection).

Each consumer uses a single channel, regardless of the number of configured queues.

Starting with version 2.0, the concurrentConsumers and maxConcurrentConsumers properties can be set with the concurrency property — for example, 2-4.

With this container, concurrency is based on the configured queues and consumersPerQueue. Each consumer for each queue uses a separate channel, and the concurrency is controlled by the rabbit client library. By default, at the time of writing, it uses a pool of DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2 threads.

You can configure a taskExecutor to provide the required maximum concurrency.

The RabbitAdmin component can declare exchanges, queues, and bindings on startup. It does this lazily, through a ConnectionListener. Consequently, if the broker is not present on startup, it does not matter. The first time a Connection is used (for example, by sending a message) the listener fires and the admin features is applied. A further benefit of doing the auto declarations in a listener is that, if the connection is dropped for any reason (for example, broker death, network glitch, and others), they are applied again when the connection is re-established.

See also RabbitMQ Automatic Connection/Topology recovery.

If you lose your connection to the broker in a synchronous sequence when using RabbitTemplate (for instance), Spring AMQP throws an AmqpException (usually, but not always, AmqpIOException). We do not try to hide the fact that there was a problem, so you have to be able to catch and respond to the exception. The easiest thing to do if you suspect that the connection was lost (and it was not your fault) is to try the operation again. You can do this manually, or you could look at using Spring Retry to handle the retry (imperatively or declaratively).

Spring Retry provides a couple of AOP interceptors and a great deal of flexibility to specify the parameters of the retry (number of attempts, exception types, backoff algorithm, and others). Spring AMQP also provides some convenience factory beans for creating Spring Retry interceptors in a convenient form for AMQP use cases, with strongly typed callback interfaces that you can use to implement custom recovery logic. See the Javadoc and properties of StatefulRetryOperationsInterceptor and StatelessRetryOperationsInterceptor for more detail. Stateless retry is appropriate if there is no transaction or if a transaction is started inside the retry callback. Note that stateless retry is simpler to configure and analyze than stateful retry, but it is not usually appropriate if there is an ongoing transaction that must be rolled back or definitely is going to roll back. A dropped connection in the middle of a transaction should have the same effect as a rollback. Consequently, for reconnections where the transaction is started higher up the stack, stateful retry is usually the best choice. Stateful retry needs a mechanism to uniquely identify a message. The simplest approach is to have the sender put a unique value in the MessageId message property. The provided message converters provide an option to do this: you can set createMessageIds to true. Otherwise, you can inject a MessageKeyGenerator implementation into the interceptor. The key generator must return a unique key for each message. In versions prior to version 2.0, a MissingMessageIdAdvice was provided. It enabled messages without a messageId property to be retried exactly once (ignoring the retry settings). This advice is no longer provided, since, along with spring-retry version 1.2, its functionality is built into the interceptor and message listener containers.

Starting with version 1.3, a builder API is provided to aid in assembling these interceptors by using Java (in @Configuration classes). The following example shows how to do so:

Only a subset of retry capabilities can be configured this way. More advanced features would need the configuration of a RetryTemplate as a Spring bean. See the Spring Retry Javadoc for complete information about available policies and their configuration.

If a MessageListener fails because of a business exception, the exception is handled by the message listener container, which then goes back to listening for another message. If the failure is caused by a dropped connection (not a business exception), the consumer that is collecting messages for the listener has to be cancelled and restarted. The SimpleMessageListenerContainer handles this seamlessly, and it leaves a log to say that the listener is being restarted. In fact, it loops endlessly, trying to restart the consumer. Only if the consumer is very badly behaved indeed will it give up. One side effect is that if the broker is down when the container starts, it keeps trying until a connection can be established.

Business exception handling, as opposed to protocol errors and dropped connections, might need more thought and some custom configuration, especially if transactions or container acks are in use. Prior to 2.8.x, RabbitMQ had no definition of dead letter behavior. Consequently, by default, a message that is rejected or rolled back because of a business exception can be redelivered endlessly. To put a limit on the client on the number of re-deliveries, one choice is a StatefulRetryOperationsInterceptor in the advice chain of the listener. The interceptor can have a recovery callback that implements a custom dead letter action — whatever is appropriate for your particular environment.

Another alternative is to set the container’s defaultRequeueRejected property to false. This causes all failed messages to be discarded. When using RabbitMQ 2.8.x or higher, this also facilitates delivering the message to a dead letter exchange.

Alternatively, you can throw a AmqpRejectAndDontRequeueException. Doing so prevents message requeuing, regardless of the setting of the defaultRequeueRejected property.

Starting with version 2.1, an ImmediateRequeueAmqpException is introduced to perform exactly the opposite logic: the message will be requeued, regardless of the setting of the defaultRequeueRejected property.

Often, a combination of both techniques is used. You can use a StatefulRetryOperationsInterceptor in the advice chain with a MessageRecoverer that throws an AmqpRejectAndDontRequeueException. The MessageRecover is called when all retries have been exhausted. The RejectAndDontRequeueRecoverer does exactly that. The default MessageRecoverer consumes the errant message and emits a WARN message.

Starting with version 1.3, a new RepublishMessageRecoverer is provided, to allow publishing of failed messages after retries are exhausted.

When a recoverer consumes the final exception, the message is ack’d and is not sent to the dead letter exchange, if any.

The following example shows how to set a RepublishMessageRecoverer as the recoverer:

The RepublishMessageRecoverer publishes the message with additional information in message headers, such as the exception message, stack trace, original exchange, and routing key. Additional headers can be added by creating a subclass and overriding additionalHeaders(). The deliveryMode (or any other properties) can also be changed in the additionalHeaders(), as the following example shows:

Starting with version 2.1, an ImmediateRequeueMessageRecoverer is added to throw an ImmediateRequeueAmqpException, which notifies a listener container to requeue the current failed message.

Spring Retry has a great deal of flexibility for determining which exceptions can invoke retry. The default configuration retries for all exceptions. Given that user exceptions are wrapped in a ListenerExecutionFailedException, we need to ensure that the classification examines the exception causes. The default classifier looks only at the top level exception.

Since Spring Retry 1.0.3, the BinaryExceptionClassifier has a property called traverseCauses (default: false). When true, it travers exception causes until it finds a match or there is no cause.

To use this classifier for retry, you can use a SimpleRetryPolicy created with the constructor that takes the max attempts, the Map of Exception instances, and the boolean (traverseCauses) and inject this policy into the RetryTemplate.

Each appender supports adding client properties to the RabbitMQ connection.

The following example shows how to add a custom client property for logback:

The properties are a comma-delimited list of key:value pairs. Keys and values cannot contain commas or colons.

These properties appear on the RabbitMQ Admin UI when the connection is viewed.

You can subclass the Logback appender. Doing so lets you modify the client connection properties before the connection is established. The following example shows how to do so:

Then you can add <thing1>thing2</thing1> to logback.xml.

For String properties such as those shown in the preceding example, the previous technique can be used. Subclasses allow for adding richer properties (such as adding a Map or numeric property).

With Log4j 2, subclasses are not supported, due to the way Log4j 2 uses static factory methods.

Within the src/main/java directory, navigate to the org.springframework.amqp.helloworld package. Open the HelloWorldConfiguration class and notice that it contains the @Configuration annotation at the class level and notice some @Bean annotations at method-level. This is an example of Spring’s Java-based configuration. You can read more about that here.

The following listing shows how the connection factory is created:

The configuration also contains an instance of RabbitAdmin, which, by default, looks for any beans of type exchange, queue, or binding and then declares them on the broker. In fact, the helloWorldQueue bean that is generated in HelloWorldConfiguration is an example because it is an instance of Queue.

The following listing shows the helloWorldQueue bean definition:

Looking back at the rabbitTemplate bean configuration, you can see that it has the name of helloWorldQueue set as its queue property (for receiving messages) and for its routingKey property (for sending messages).

Now that we have explored the configuration, we can look at the code that actually uses these components. First, open the Producer class from within the same package. It contains a main() method where the Spring ApplicationContext is created.

The following listing shows the main method:

In the preceding example, the AmqpTemplate bean is retrieved and used for sending a Message. Since the client code should rely on interfaces whenever possible, the type is AmqpTemplate rather than RabbitTemplate. Even though the bean created in HelloWorldConfiguration is an instance of RabbitTemplate, relying on the interface means that this code is more portable (you can change the configuration independently of the code). Since the convertAndSend() method is invoked, the template delegates to its MessageConverter instance. In this case, it uses the default SimpleMessageConverter, but a different implementation could be provided to the rabbitTemplate bean, as defined in HelloWorldConfiguration.

Now open the Consumer class. It actually shares the same configuration base class, which means it shares the rabbitTemplate bean. That is why we configured that template with both a routingKey (for sending) and a queue (for receiving). As we describe in AmqpTemplate, you could instead pass the 'routingKey' argument to the send method and the 'queue' argument to the receive method. The Consumer code is basically a mirror image of the Producer, calling receiveAndConvert() rather than convertAndSend().

The following listing shows the main method for the Consumer:

If you run the Producer and then run the Consumer, you should see Received: Hello World in the console output.

Synchronous Example walked through the synchronous Hello World sample. This section describes a slightly more advanced but significantly more powerful option. With a few modifications, the Hello World sample can provide an example of asynchronous reception, also known as message-driven POJOs. In fact, there is a sub-package that provides exactly that: org.springframework.amqp.samples.helloworld.async.

Again, we start with the sending side. Open the ProducerConfiguration class and notice that it creates a connectionFactory and a rabbitTemplate bean. This time, since the configuration is dedicated to the message sending side, we do not even need any queue definitions, and the RabbitTemplate has only the 'routingKey' property set. Recall that messages are sent to an exchange rather than being sent directly to a queue. The AMQP default exchange is a direct exchange with no name. All queues are bound to that default exchange with their name as the routing key. That is why we only need to provide the routing key here.

The following listing shows the rabbitTemplate defintion:

Since this sample demonstrates asynchronous message reception, the producing side is designed to continuously send messages (if it were a message-per-execution model like the synchronous version, it would not be quite so obvious that it is, in fact, a message-driven consumer). The component responsible for continuously sending messages is defined as an inner class within the ProducerConfiguration. It is configured to run every three seconds.

The following listing shows the component:

You do not need to understand all of the details, since the real focus should be on the receiving side (which we cover next). However, if you are not yet familiar with Spring task scheduling support, you can learn more here. The short story is that the postProcessor bean in the ProducerConfiguration registers the task with a scheduler.

Now we can turn to the receiving side. To emphasize the message-driven POJO behavior, we start with the component that react to the messages. The class is called HelloWorldHandler and is shown in the following listing:

That class is a POJO. It does not extend any base class, it does not implement any interfaces, and it does not even contain any imports. It is being “adapted” to the MessageListener interface by the Spring AMQP MessageListenerAdapter. You can then configure that adapter on a SimpleMessageListenerContainer. For this sample, the container is created in the ConsumerConfiguration class. You can see the POJO wrapped in the adapter there.

The following listing shows how the listenerContainer is defined:

The SimpleMessageListenerContainer is a Spring lifecycle component and, by default, starts automatically. If you look in the Consumer class, you can see that its main() method consists of nothing more than a one-line bootstrap to create the ApplicationContext. The Producer’s main() method is also a one-line bootstrap, since the component whose method is annotated with @Scheduled also starts automatically. You can start the Producer and Consumer in any order, and you should see messages being sent and received every three seconds.

BrokerRunning provides a mechanism to let tests succeed when a broker is not running (on localhost, by default).

It also has utility methods to initialize and empty queues and delete queues and exchanges.

The following example shows its usage:

There are several isRunning…​ static methods, such as isBrokerAndManagementRunning(), which verifies the broker has the management plugin enabled.

LongRunningIntegrationTest is a rule that disables long running tests. You might want to use this on a developer system but ensure that the rule is disabled on, for example, nightly CI builds.

The following example shows its usage:

To disable the rule at runtime, set an environment variable called RUN_LONG_INTEGRATION_TESTS to true.

This class-level annotation is similar to the BrokerRunning @Rule discussed in JUnit4 @Rules. It is processed by the RabbitAvailableCondition.

The annotation has two properties:

It is used to check whether the broker is available and skip the tests if not. As discussed in Configuring the Rule, the environment variable called RABBITMQ_SERVER_REQUIRED, if true, causes the tests to fail fast if there is no broker. You can configure the condition by using environment variables as discussed in Configuring the Rule.

In addition, the RabbitAvailableCondition supports argument resolution for parameterized test constructors and methods. Two argument types are supported:

The following example shows both:

The preceding test is in the framework itself and verifies the argument injection and that the condition created the queue properly.

A practical user test might be as follows:

When you use a Spring annotation application context within a test class, you can get a reference to the condition’s connection factory through a static method called RabbitAvailableCondition.getBrokerRunning(). The following test comes from the framework and demonstrates the usage:

Similar to the LongRunningIntegrationTest JUnit4 @Rule, this annotation causes tests to be skipped unless an environment variable (or system property) is set to true. The following example shows how to use it:

By default, the variable is RUN_LONG_INTEGRATION_TESTS, but you can specify the variable name in the annotation’s value attribute.