The Spring AMQP project consists of a handful of modules, each represented by a JAR in the distribution. These modules are: spring-amqp, spring-rabbit, spring-erlang, and spring-rabbit-admin. The 'spring-amqp' module contains the org.springframework.amqp.core package. Within that package, you will find the classes that represent the core AMQP "model". Our intention is to provide generic abstractions that do not rely on any particular AMQP broker implementation or client library. End user code will be more portable across vendor implementations as it can be developed against the abstraction layer only. These abstractions are then used implemented by broker-specific modules, such as 'spring-rabbit'. For the M1 release there is only a RabbitMQ implementation however the abstractions have been vetted in .NET using Apache Qpid in addition to RabbitMQ.

The overview here assumes that you are already familiar with the basics of the AMQP specification already. If you are not, then have a look at the resources listed in Part III, “Other Resources”

public class Message {

    private final MessageProperties messageProperties;

    private final byte[] body;

    public Message(byte[] body, MessageProperties messageProperties) {
        this.body = body;
        this.messageProperties = messageProperties;
    }

    public byte[] getBody() {
        return this.body;
    }

    public MessageProperties getMessageProperties() {
        return this.messageProperties;
}

public interface Exchange {

    String getName();

    ExchangeType getExchangeType();

    boolean isDurable();

    boolean isAutoDelete();

    Map<String, Object> getArguments();

}

public class Queue  {

    private final String name;

    private volatile boolean durable;

    private volatile boolean exclusive;

    private volatile boolean autoDelete;

    private volatile Map<String, Object> arguments;

    public Queue(String name) {
        this.name = name;
    }

    // Getters and Setters omitted for brevity

new Binding(someQueue, someDirectExchange, "foo.bar")

new Binding(someQueue, someTopicExchange, "foo.*")

new Binding(someQueue, someFanoutExchange)

Binding b = BindingBuilder.from(someQueue).to(someTopicExchange).with("foo.*");

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 Section 1.3, “AmqpTemplate”).

Whereas the AMQP model we described in the previous section is generic and applicable to all implementations, when we get into the management of resources, the details are specific to the broker implementation. Therefore, in this section, we will be focusing on code that exists only within our "spring-rabbit" module since at this point, RabbitMQ is the only supported implementation.

The central component for managing a connection to the RabbitMQ broker is the ConnectionFactory interface. The responsibility of a ConnectionFactory implementation is to provide an instance of com.rabbitmq.client.Connection. The simplest implementation we provide is SingleConnectionFactory which establishes a single connection that can be shared by the application. Sharing of the connection is possible since the "unit of work" for messaging with AMQP is actually a "channel" (in some ways, this is similar to the relationship between a Connection and a Session in JMS). As you can imagine, the connection instance provides a createChannel method. When creating an instance of SingleConnectionFactory, the 'hostname' can be provided via the constructor. The 'username' and 'password' properties should be provided as well.

SingleConnectionFactory connectionFactory = new SingleConnectionFactory("somehost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");

Connection connection = connectionFactory.createConnection();

When using XML, the configuration might look like this:

<bean id="cf" class="org.springframework.amqp.rabbit.connection.SingleConnectionFactory">
    <constructor-arg value="somehost"/>
    <property name="username" value="guest"/>
    <property name="password" value="guest"/>
</bean>

As with many other high-level abstractions provided by the Spring Framework and related projects, Spring AMQP provides a "template" that plays a central role. The interface that defines the main operations is called AmqpTemplate. Those operations cover the general behavior for sending and receiving Messages. In other words, they are not unique to any implementation, hence the "AMQP" in the name. On the other hand, there are implementations of that interface that are tied to implementations of the AMQP protocol. Unlike JMS, which is an interface-level API itself, AMQP is a wire-level protocol. The implementations of that protocol provide their own client libraries, so each implementation of the template interface will depend on a particular client library. Currently, there is only a single implementation: RabbitTemplate. In the examples that follow, you will often see usage of an "AmqpTemplate", but when you look at the configuration examples, or any code excerpts where the template is instantiated and/or setters are invoked, you will see the implementation type (e.g. "RabbitTemplate").

As mentioned above, the AmqpTemplate interface defines all of the basic operations for sending and receiving Messages. We will explore Message sending and reception, respectively, in the two sections that follow.

When sending a Message, one can use any of the following methods:

void send(MessageCreator messageCreator) throws AmqpException;

void send(String routingKey, MessageCreator messageCreator) throws AmqpException;

void send(String exchange, String routingKey, MessageCreator messageCreator) throws AmqpException;

We can begin our discussion with the last method listed above since it is actually the most explicit. It allows an AMQP Exchange name to be provided at runtime along with a routing key. The last parameter is the callback that is responsible for actual creating of the Message instance. An example of using this method to send a Message might look this this:

amqpTemplate.send("marketData.topic", "quotes.nasdaq.FOO", new MessageCreator() {
    public Message createMessage() {
        return new Message("12.34".getBytes(), someProperties);
    }
});

The "exchange" property can be set on the template itself if you plan to use that template instance to send to the same exchange most or all of the time. In such cases, the second method listed above may be used instead. The following example is functionally equivalent to the previous one:

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.send("quotes.nasdaq.FOO", new MessageCreator() {
    public Message createMessage() {
        return new Message("12.34".getBytes(), someProperties);
    }
});

If both the "exchange" and "routingKey" properties are set on the template, then the method accepting only the MessageCreator may be used:

amqpTemplate.setExchange("marketData.topic");
amqpTemplate.setRoutingKey("quotes.nasdaq.FOO");
amqpTemplate.send(new MessageCreator() {
    public Message createMessage() {
        return new Message("12.34".getBytes(), someProperties);
    }
});

A better way of thinking about the exchange and routing key properties is that the explicit method parameters will always override the template's default values. In fact, even if you do not explicitly set those properties on the template, there are always default values in place. In both cases, the default is an empty String, but that is actually a sensible default. As far as the routing key is concerned, it's not always necessary in the first place (e.g. a Fanout Exchange). Furthermore, a Queue may be bound to an Exchange with an empty String. Those are both legitimate scenarios for reliance on the default empty String value for the routing key property of the template. As far as the Exchange name is concerned, the empty String is quite commonly used because the AMQP specification defines the "default Exchange" as having no name. Since all Queues are automatically bound to that default Exchange (which is a Direct Exchange) using their name as the binding value, that second method above can be used for simple point-to-point Messaging to any Queue through the default Exchange. Simply provide the queue name as the "routingKey" - either by providing the method parameter at runtime:

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.send("queue.helloWorld", new MessageCreator() {
    public Message createMessage() {
        return new Message("Hello World".getBytes(), someProperties);
    }
});

Or, if you prefer to create a template that will be used for publishing primarily or exclusively to a single Queue, the following is perfectly reasonable:

RabbitTemplate template = new RabbitTemplate(); // using default no-name Exchange
template.setRoutingKey("queue.helloWorld"); // but we'll always send to this Queue
template.send(new MessageCreator() {
    public Message createMessage() {
        return new Message("Hello World".getBytes(), someProperties);
    }
});

Message reception is always a bit more complicated than sending. The reason is that there are two ways to receive a Message. The simpler option is to poll for a single Message at a time with a synchronous, blocking method call. The more complicated yet more common approach is to register a listener that will receive Messages on-demand, asynchronously. We will look at an example of each approach in the next two sub-sections.

Message receive() throws AmqpException;

Message receive(String queueName) throws AmqpException;

public interface MessageListener {

    void onMessage(Message message);

}

public interface ChannelAwareMessageListener<M extends Message> {

    void onMessage(M message, Channel channel) throws Exception;

}

MessageListener listener = new MessageListenerAdapter(somePojo);

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(rabbitConnectionFactory);
container.setQueueName("some.queue");
container.setMessageListener(someListener);

<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
    <property name="connectionFactory" ref="rabbitConnectionFactory"/>
    <property name="queueName" value="some.queue"/>
    <property name="messageListener" ref="someListener"/>
</bean>

@Configuration
public class ExampleAmqpConfiguration {

    @Bean
    public MessageListenerContainer messageListenerContainer {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory());
        container.setQueueName("some.queue");
        container.setMessageListener(exampleListener());
        return container;
    }

    @Bean
    public ConnectionFactory rabbitConnectionFactory() {
        SingleConnectionFactory connectionFactory = new SingleConnectionFactory("localhost");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }

    @Bean
    public MessageListener exampleListener() {
        return new MessageListener() {
            public void onMessage(Message message) {
                System.out.println("received: " + message)
            }
        }
    }
}

The AmqpTemplate also defines several methods for sending and receiving Messages that will delegate to a MessageConverter. The MessageConverter itself is quite straightforward. It provides a single method for each direction: one for converting to a Message and another for converting from a Message. Notice that when converting to a Message, you may also provide properties in addition to the object. The "object" parameter typically corresponds to the Message body.

public interface MessageConverter {

    Message toMessage(Object object, MessageProperties messageProperties)
            throws MessageConversionException;

    Object fromMessage(Message message) throws MessageConversionException;

}

The relevant Message-sending methods on the AmqpTemplate are listed below. They are simpler than the methods we discussed previously because they do not require the MessageCreator callback. Instead, the MessageConverter is responsible for "creating" each Message by converting the provided object to the byte array for the Message body and then adding any provided MessageProperties.

void convertAndSend(Object message) throws AmqpException;

void convertAndSend(String routingKey, Object message) throws AmqpException;

void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;

void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException;

void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor)
        throws AmqpException;

void convertAndSend(String exchange, String routingKey, Object message,
        MessagePostProcessor messagePostProcessor) throws AmqpException;

On the receiving side, there are only two methods: one that accepts the queue name and one that relies on the template's "queue" property having been set.

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <property name="connectionFactory" ref="rabbitConnectionFactory"/>
    <property name="messageConverter">
        <bean class="org.springframework.amqp.support.converter.JsonMessageConverter">
            <!-- if necessary, override the DefaultClassMapper -->
            <property name="classMapper" ref="customClassMapper"/>
        </bean>
    </property>
</bean>

<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate">
    <property name="connectionFactory" ref="rabbitConnectionFactory"/>
    <property name="messageConverter">
        <bean class="org.springframework.amqp.support.converter.MarshallingMessageConverter">
            <constructor-arg ref="someImplemenationOfMarshallerAndUnmarshaller"/>
        </bean>
    </property>
</bean>

The AMQP specification describes how the protocol can be used to configure Queues, Exchanges and Bindings on the broker. These operations which are portable from the 0.8 specification and higher are present in the AmqpAdmin interface in the org.springframework.amqp.core package. The RabbitMQ implementation of that class is RabbitAdmin located in the org.springframework.amqp.rabbit.core package. Any many configuration and management functions are broker specific and not included in the AMQP specification, the interface RabbitBrokerOperations and its implementation RabbitBrokerAdmin located in the org.springframework.amqp.rabbit.admin package is provided to fill that gap.

The AmqpAdmin interface is based on using the Spring AMQP domain abstractions and is shown below:

public interface AmqpAdmin {

    // Exchange Operations

    void declareExchange(Exchange exchange);

    void deleteExchange(String exchangeName);

    // Queue Operations

    Queue declareQueue();

    void declareQueue(Queue queue);

    void deleteQueue(String queueName);

    void deleteQueue(String queueName, boolean unused, boolean empty);

    void purgeQueue(String queueName, boolean noWait);

    // Binding Operations

    void declareBinding(Binding binding);

}

The declareQueue() method defined a queue on the broker whose name is automatically created. The additional properties of this auto-generated queue are exclusive=true, autoDelete=true, and durable=false.

The RabbitMQ implementation of this interface is RabbitAdmin which when configured using Spring XML would look like this:

<bean id="cf" class="org.springframework.amqp.rabbit.connection.SingleConnectionFactory">
    <constructor-arg value="localhost"/>
    <property name="username" value="guest"/>
    <property name="password" value="guest"/>
</bean>

<bean id="amqpAdmin" class="org.springframework.amqp.rabbit.core.RabbitAdmin">
    <constructor-arg ref="cf"/>
</bean>

The class AbstractAmqpConfiguration can also be used to configure an instance of the AmqpAmin interface. The base class AbstractAmqpConfiguration is located in the package org.springframework.amqp.config and is shown in part below.

@Configuration
public abstract class AbstractAmqpConfiguration implements ApplicationContextAware, SmartLifecycle {

  @Bean
  public abstract AmqpAdmin amqpAdmin();

  // the rest omitted for brevity.
}

The part that is omitted is related to the implementation of SmartLifecycle which is responsible for querying the container for all Queues, Exchanges, and Bindings that are defined in the DI container and declaring them to the broker. Thus if you are using the @Configuration style of configuration, you can simply create Queue, Exchange, and Binding bean definitions and they will be declared to the broker once the application has started.

If you are using RabbitMQ as your broker, the additional abstract @Configuration class can be used to bootstrap an implementation of AmqpAdmin as shown below

@Configuration
public abstract class AbstractRabbitConfiguration extends AbstractAmqpConfiguration {

  @Bean
  public abstract RabbitTemplate rabbitTemplate();

  @Bean
  public AmqpAdmin amqpAdmin() {
    this.amqpAdmin = new RabbitAdmin(rabbitTemplate().getConnectionFactory());
    return this.amqpAdmin;
  }
}

This leaves it up to you to provide an implementation of the rabbitTemplate method in your application specific subclass. For example, using the Stock sample application, there is the @Configuration class AbstractStockRabbitConfiguration which in turn has RabbitClientConfiguration and RabbitServerConfiguration subclasses. The code for AbstractStockRabbitConfiguration is show below

@Configuration
public abstract class AbstractStockAppRabbitConfiguration extends AbstractRabbitConfiguration {

    @Bean
    public ConnectionFactory connectionFactory() {
        SingleConnectionFactory connectionFactory = new SingleConnectionFactory("localhost");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMessageConverter(jsonMessageConverter());
        configureRabbitTemplate(template);
        return template;
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new JsonMessageConverter();
    }

    @Bean
    public TopicExchange marketDataExchange() {
        return new TopicExchange("app.stock.marketdata");
    }

    // additional code omitted for brevity

}

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

@Configuration
public class RabbitServerConfiguration extends AbstractStockAppRabbitConfiguration  {

    @Bean
    public Queue stockRequestQueue() {
        return new Queue("app.stock.request");
    }
}

This is the end of the whole inheritance chain of @Configuration classes. The end result is the the TopicExchange and Queue will be declared to the broker upon application startup. There is no binding of the 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 and is show below.

@Configuration
public class RabbitClientConfiguration extends AbstractStockAppRabbitConfiguration {

    @Value("${stocks.quote.pattern}")
    private String marketDataRoutingKey;

    @Bean
    public Queue marketDataQueue() {
        return amqpAdmin().declareQueue();
    }

    /**
     * Binds to the market data exchange. Interested in any stock quotes.
     */
    @Bean
    public Binding marketDataBinding() {
        return BindingBuilder.from(
                marketDataQueue()).to(marketDataExchange()).with(marketDataRoutingKey);
    }

    // additional code omitted for brevity

}

The client is declaring another queue via the declareQueue() method on the AmqpAdmin, and it binds that queue to the market data exchange with a routing pattern that is externalized in a properties file.

Many operations with the RabbitMQ Java client can throw checked Exceptions. For example, there are a lot of cases where IOExceptions may be thrown. The RabbitTemplate, SimpleMessageListenerContainer, and other Spring AMQP components will catch those Exceptions and convert into one of the Exceptions within our runtime hierarchy. Those are defined in the 'org.springframework.amqp' package, and AmqpException is the base of the hierarchy.

The Spring Rabbit framework has support for automatic transaction management in the synchronous and asynchronous use cases with a number of different semantics that can be selected declaratively, as is familiar to existing users of Spring transactions. This makes many if not most common messaging patterns very easy to implement.

There are two ways to signal the desired transaction semantics to the framework. In both the RabbitTemplate and SimpleMessageListenerContainer there is a flag channelTransacted which, if true, tells the framework to use a transactional channel and to end all operations (send or receive) with a commit or rollback depending on the outcome, with an exception signalling a rollback. Another signal is to provide an external transaction with one of Spring's PlatformTransactionManager implementations as a context for the ongoing operation. If there is already a transaction in progress when the framework is sending or receiving a message, and the channelTransacted flag is true, then the commit or rollback of the messaging transaction will be deferred until the end of the current transaction. If the channelTransacted flag is false, then no transaction semantics apply to the messaging operation (it is auto-acked).

The channelTransacted flag is a configuration time setting: it is declared and processed once when the AMQP components are created, usually at application startup. The external transaction is more dynamic in principle because the system responds to the current Thread state at runtime, but in practice is often also a configuration setting, when the transactions are layed onto an application declaratively.

For synchronous use cases with RabbitTemplate the external transaction is provided by the caller, either declaratively or imperatively according to taste (the usual Spring transaction model). An example of a declarative approach (usually preferred because it is non-invasive), where the template has been configured with channelTransacted=true:

@Transactional
public void doSomething() {
    String incoming = rabbitTemplate.receiveAndConvert();
    // do some more database processing...
    String outgoing = processInDatabaseAndExtractReply(incoming);
    rabbitTemplate.convertAndSend(outgoing);
}

A String payload is received, converted and sent as a message body inside a method marked as @Transactional, so if the database processing fails with an exception, the incoming message will be returned to the broker, and the outgoing message will not be sent. This applies to any operations with the RabbitTemplate inside a chain of transactional methods (unless the Channel is directly manipulated to commit the transactiom early for instance).

For asynchronous use cases with SimpleMessageListenerContainer an external transaction has to be requested by the container when it sets up the listener. To signal that an external transaction is required the user provides an implementation of PlatformTransactionManager to the container when it is configured. For example:

@Configuration
public class ExampleExternalTransactionAmqpConfiguration {

    @Bean
    public MessageListenerContainer messageListenerContainer {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory());
        container.setTransactionManager(transactionManager());
        container.setChannelTransacted(true);
        container.setQueueName("some.queue");
        container.setMessageListener(exampleListener());
        return container;
    }

}

In the example above, the transaction manager is added as a dependency injected from another bean definition (not shown), and the channelTransacted flag is also set to true. The effect is that if the listener fails with an exception the transaction will be rolled back, and the message will also be returned to the broker. Significantly, if the transaction fails to commit (e.g. a database constraint error, or connectivity problem), then the AMQP transaction will also be rolled back, and message will be returned to the broker. This is sometimes known as a Best Efforts 1 Phase Commit, and is a very powerful pattern for reliable messaging. If the channelTransacted flag was set to false in the example above, which is the default, then the external transaction would still be provided for the listener, but all messaging operations would be auto-acked, so the effect is to commit the messaging operations even on a rollback of the business operation.

A note on rollback of received messages. AMQP transactions only apply to messages and acks sent to the broker, so when there is a rollback of a Spring transaction and a message has been received, wheat Spring AMQP has to do is not just rollback the transaction, but also manually reject the message (sort of a nack, but that's not what the specification calls it). Such messages (and any that are unacked when a channel is closed or aborts) go to the back of the queue on a Rabbit broker, and this behaviour is not what some users expect, especially if they come from a JMS background, so it's good to be aware of it. The re-queuing order is not mandated by the AMQP specification, but it makes the broker much more efficient, and also means that if it is under load there is a natural back off before the message can be consumed again.

All the methods in the RabbitBrokerAdmin class are exposed through JMX through the use of the @ManagedOperations annotation. Please refer to the sample application for how to configure spring to export the RabbitBrokerAdmin.