This version is still in development and is not considered stable yet. For the latest stable version, please use Spring AMQP 3.1.6! |
Sample Applications
The Spring AMQP Samples project includes two sample applications. The first is a simple “Hello World” example that demonstrates both synchronous and asynchronous message reception. It provides an excellent starting point for acquiring an understanding of the essential components. The second sample is based on a stock-trading use case to demonstrate the types of interaction that would be common in real world applications. In this chapter, we provide a quick walk-through of each sample so that you can focus on the most important components. The samples are both Maven-based, so you should be able to import them directly into any Maven-aware IDE (such as SpringSource Tool Suite).
The “Hello World” Sample
The “Hello World” sample demonstrates both synchronous and asynchronous message reception.
You can import the spring-rabbit-helloworld
sample into the IDE and then follow the discussion below.
Synchronous Example
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:
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory =
new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}
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:
@Bean
public Queue helloWorldQueue() {
return new Queue(this.helloWorldQueueName);
}
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:
public static void main(String[] args) {
ApplicationContext context =
new AnnotationConfigApplicationContext(RabbitConfiguration.class);
AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
amqpTemplate.convertAndSend("Hello World");
System.out.println("Sent: Hello World");
}
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
:
public static void main(String[] args) {
ApplicationContext context =
new AnnotationConfigApplicationContext(RabbitConfiguration.class);
AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
System.out.println("Received: " + amqpTemplate.receiveAndConvert());
}
If you run the Producer
and then run the Consumer
, you should see Received: Hello World
in the console output.
Asynchronous Example
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
definition:
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey(this.helloWorldQueueName);
return template;
}
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:
static class ScheduledProducer {
@Autowired
private volatile RabbitTemplate rabbitTemplate;
private final AtomicInteger counter = new AtomicInteger();
@Scheduled(fixedRate = 3000)
public void sendMessage() {
rabbitTemplate.convertAndSend("Hello World " + counter.incrementAndGet());
}
}
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:
public class HelloWorldHandler {
public void handleMessage(String text) {
System.out.println("Received: " + text);
}
}
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:
@Bean
public SimpleMessageListenerContainer listenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueName(this.helloWorldQueueName);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
return container;
}
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.
Stock Trading
The Stock Trading sample demonstrates more advanced messaging scenarios than the Hello World sample.
However, the configuration is very similar, if a bit more involved.
Since we walked through the Hello World configuration in detail, here, we focus on what makes this sample different.
There is a server that pushes market data (stock quotations) to a topic exchange.
Then, clients can subscribe to the market data feed by binding a queue with a routing pattern (for example,
app.stock.quotes.nasdaq.*
).
The other main feature of this demo is a request-reply “stock trade” interaction that is initiated by the client and handled by the server.
That involves a private replyTo
queue that is sent by the client within the order request message itself.
The server’s core configuration is in the RabbitServerConfiguration
class within the org.springframework.amqp.rabbit.stocks.config.server
package.
It extends the AbstractStockAppRabbitConfiguration
.
That is where the resources common to the server and client are defined, including the market data topic exchange (whose name is 'app.stock.marketdata') and the queue that the server exposes for stock trades (whose name is 'app.stock.request').
In that common configuration file, you also see that a Jackson2JsonMessageConverter
is configured on the RabbitTemplate
.
The server-specific configuration consists of two things.
First, it configures the market data exchange on the RabbitTemplate
so that it does not need to provide that exchange name with every call to send a Message
.
It does this within an abstract callback method defined in the base configuration class.
The following listing shows that method:
public void configureRabbitTemplate(RabbitTemplate rabbitTemplate) {
rabbitTemplate.setExchange(MARKET_DATA_EXCHANGE_NAME);
}
Second, the stock request queue is declared.
It does not require any explicit bindings in this case, because it is bound to the default no-name exchange with its own name as the routing key.
As mentioned earlier, the AMQP specification defines that behavior.
The following listing shows the definition of the stockRequestQueue
bean:
@Bean
public Queue stockRequestQueue() {
return new Queue(STOCK_REQUEST_QUEUE_NAME);
}
Now that you have seen the configuration of the server’s AMQP resources, navigate to the org.springframework.amqp.rabbit.stocks
package under the src/test/java
directory.
There, you can see the actual Server
class that provides a main()
method.
It creates an ApplicationContext
based on the server-bootstrap.xml
config file.
There, you can see the scheduled task that publishes dummy market data.
That configuration relies upon Spring’s task
namespace support.
The bootstrap config file also imports a few other files.
The most interesting one is server-messaging.xml
, which is directly under src/main/resources
.
There, you can see the messageListenerContainer
bean that is responsible for handling the stock trade requests.
Finally, have a look at the serverHandler
bean that is defined in server-handlers.xml
(which is also in 'src/main/resources').
That bean is an instance of the ServerHandler
class and is a good example of a message-driven POJO that can also send reply messages.
Notice that it is not itself coupled to the framework or any of the AMQP concepts.
It accepts a TradeRequest
and returns a TradeResponse
.
The following listing shows the definition of the handleMessage
method:
public TradeResponse handleMessage(TradeRequest tradeRequest) { ...
}
Now that we have seen the most important configuration and code for the server, we can turn to the client.
The best starting point is probably RabbitClientConfiguration
, in the org.springframework.amqp.rabbit.stocks.config.client
package.
Notice that it declares two queues without providing explicit names.
The following listing shows the bean definitions for the two queues:
@Bean
public Queue marketDataQueue() {
return amqpAdmin().declareQueue();
}
@Bean
public Queue traderJoeQueue() {
return amqpAdmin().declareQueue();
}
Those are private queues, and unique names are generated automatically.
The first generated queue is used by the client to bind to the market data exchange that has been exposed by the server.
Recall that, in AMQP, consumers interact with queues while producers interact with exchanges.
The “binding” of queues to exchanges is what tells the broker to deliver (or route) messages from a given exchange to a queue.
Since the market data exchange is a topic exchange, the binding can be expressed with a routing pattern.
The RabbitClientConfiguration
does so with a Binding
object, and that object is generated with the BindingBuilder
fluent API.
The following listing shows the Binding
:
@Value("${stocks.quote.pattern}")
private String marketDataRoutingKey;
@Bean
public Binding marketDataBinding() {
return BindingBuilder.bind(
marketDataQueue()).to(marketDataExchange()).with(marketDataRoutingKey);
}
Notice that the actual value has been externalized in a properties file (client.properties
under src/main/resources
), and that we use Spring’s @Value
annotation to inject that value.
This is generally a good idea.
Otherwise, the value would have been hardcoded in a class and unmodifiable without recompilation.
In this case, it is much easier to run multiple versions of the client while making changes to the routing pattern used for binding.
We can try that now.
Start by running org.springframework.amqp.rabbit.stocks.Server
and then org.springframework.amqp.rabbit.stocks.Client
.
You should see dummy quotations for NASDAQ
stocks, because the current value associated with the 'stocks.quote.pattern' key in client.properties is 'app.stock.quotes.nasdaq.'.
Now, while keeping the existing Server
and Client
running, change that property value to 'app.stock.quotes.nyse.' and start a second Client
instance.
You should see that the first client still receives NASDAQ quotes while the second client receives NYSE quotes.
You could instead change the pattern to get all stocks or even an individual ticker.
The final feature we explore is the request-reply interaction from the client’s perspective.
Recall that we have already seen the ServerHandler
that accepts TradeRequest
objects and returns TradeResponse
objects.
The corresponding code on the Client
side is RabbitStockServiceGateway
in the org.springframework.amqp.rabbit.stocks.gateway
package.
It delegates to the RabbitTemplate
in order to send messages.
The following listing shows the send
method:
public void send(TradeRequest tradeRequest) {
getRabbitTemplate().convertAndSend(tradeRequest, new MessagePostProcessor() {
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setReplyTo(new Address(defaultReplyToQueue));
try {
message.getMessageProperties().setCorrelationId(
UUID.randomUUID().toString().getBytes("UTF-8"));
}
catch (UnsupportedEncodingException e) {
throw new AmqpException(e);
}
return message;
}
});
}
Notice that, prior to sending the message, it sets the replyTo
address.
It provides the queue that was generated by the traderJoeQueue
bean definition (shown earlier).
The following listing shows the @Bean
definition for the StockServiceGateway
class itself:
@Bean
public StockServiceGateway stockServiceGateway() {
RabbitStockServiceGateway gateway = new RabbitStockServiceGateway();
gateway.setRabbitTemplate(rabbitTemplate());
gateway.setDefaultReplyToQueue(traderJoeQueue());
return gateway;
}
If you are no longer running the server and client, start them now. Try sending a request with the format of '100 TCKR'. After a brief artificial delay that simulates “processing” of the request, you should see a confirmation message appear on the client.
Receiving JSON from Non-Spring Applications
Spring applications, when sending JSON, set the TypeId
header to the fully qualified class name to assist the receiving application in converting the JSON back to a Java object.
The spring-rabbit-json
sample explores several techniques to convert the JSON from a non-Spring application.
See also Jackson2JsonMessageConverter
as well as the Javadoc for the DefaultClassMapper
.