Tuesday, July 23, 2013

Introduction to messaging with Spring JMS

1   Introduction

In this post I will show you how to configure a standalone application in order to see different ways of sending and receiving messages using Spring JMS. Basically, I will divide the examples into the following sections:
  • Point-to-point messaging (queue)
    • Synchronous reception
    • Asynchronous reception
  • Publish-subscribe messaging (topic)

The source code with all the examples shown in this article is available at github.


2   Configuring the provider

The first thing we need to do is to configure the ConnectionFactory. The connection factory is part of the JMS specification and allows the application to create connections with the JMS provider.

The factory is used to create a connection which is then used to create a session. In the following examples, we won't need to care about this since the JmsTemplate class will do this for us.

Spring provides its own ConnectionFactory implementations, which are specified below:
  • SingleConnectionFactory: All createConnection() calls will return the same connection. This is useful for testing.
  • CachingConnectionFactory: It provides caching of sessions.

The JmsTemplate aggressively opens and closes resources like sessions since it assumes that are cached by the connectionFactory. Using the CachingConnectionFactory will improve its performance. In our example, we will define a cachingConnectionFactory passing our previously defined AMQ connectionFactory to its targetConnectionFactory property:


3   Point-to-point messaging (queue)

This Destination implementation consists in sending a message to a single consumer. The producer will send a message to the queue where it will be retrieved by the consumer.



The consumer will actively retrieve the message from the queue (synchronous reception) or it will retrieve the message passively (asynchronous reception). Now we will see an example of each.


3.1   Synchronous reception


3.1.1   Configuration


Spring JMS uses JmsTemplate class for message production and synchronous message reception. This template is a central class of Spring JMS, and helps us by:
  • Reducing boilerplate code: It handles the creation and release of resources transparently (connection, session...).
  • Handling exceptions and converting them to runtime exceptions.
  • Providing utility methods and callbacks.

Let's configure the jmsTemplate:

This template will handle our point-to-point messaging. To use topics it will need further configuration, which will be shown in the following sections.

The Queue destination is defined below:

In our example, a producer will send a message to this queue and a consumer will retrieve it.

3.1.2   The producer


This producer will use a jmsTemplate to send a message. Note the @Component annotation, this class will be auto detected and registered as a bean.

It is also important to see that we are passing a Notification object to the jmsTemplate method. If we do not define a message converter, the template will register a SimpleMessageConverter by default (check JmsTemplate constructor). This converter will be able to convert the following types:
  • String to TextMessage
  • Serializable to ObjectMessage
  • Map to MapMessage
  • byte[] to BytesMessage

If the object being sent is not an instance of any of the previous list, it will throw a MessageConversionException. The common cause of this exception is that your object is not implementing Serializable interface.

In this case, it will convert our Notification object to an ObjectMessage and send it to its default destination, which we previously defined as "test.sync.queue".

3.1.3   The consumer


You should use this method carefully as it blocks the current thread until it receives the message. You should better define a timeout in case there's a problem receiving the message. The jmsTemplate has no timeout setter method defined. You will need to define it when configuring the AMQ connection factory:

<property name="sendTimeout" value="5000"/>

3.1.4   The test



3.2   Asynchronous reception


Spring lets you receive messages asynchronously in two different ways:
  • Implementing MessageListener interface
  • Using a simple POJO

The following example will show the second approach.

3.2.1   Configuration


We can use the same jmsTemplate we configured in the previous example. In this case we will configure another queue where the producer will send its message and a consumer that will act as a listener:

We are configuring a consumer which will be the asyncReceiver bean, and the listener container will invoke its receiveMessage method when a message arrives to the test.async.queue.

3.2.2   The producer


The producer will be the same defined in the previous section, but it will send the message to a different queue:


3.2.3   The consumer


As you can see, it's a simple Java class. It does not need to implement any interface. The consumer saves received notifications to a registry. This registry will be used by the test class to assert that notifications arrived correctly.

3.2.4   The test



4   Publish-subscribe messaging (topic)

The message is sent to a topic, where it will be distributed to all consumers that are subscribed to this topic.



4.1   Configuration


We will need another jmsTemplate since the template we configured before is set to work with queues.

We just need to configure its destination accessor by defining its pubSubDomain property and set its value to true. The default value is false (point-to-point).

Next, we configure a new destination, which will be the topic for this example:

Finally, we define the listeners. We define two listeners to make sure both consumers receive the message sent to the topic by the producer.

You may notice a difference in the listener configuration. We need to change the destination type of the listener container, which is set to queue by default. Just set its value to topic and we are done.

4.2   The producer



4.3   The consumer


The asyncTopicFooReceiver has the same method.

4.4   The test


19 comments:

  1. How do i add programmatically or dynamically suscribers to the JMS topic? can i reference the topic with the session object and add listeners?

    ReplyDelete
  2. You can create a dynamic subscriber programmatically from the session, which is created from your injected connection factory:

    Connection con = connectionFactory.createConnection();
    Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageConsumer consumer = session.createConsumer(destination);

    Then you just need to define a listener for the consumer you've just created.

    I've added a test case named TestDynamicTopicReceiver to my source code. You can check it here:

    https://github.com/xpadro/spring-integration/tree/master/jms-basic

    ReplyDelete
    Replies
    1. Thanks Xavier for this great blog!
      Just another question about this dynamic subscriber code,
      Where this destination @Autowired from? I don't see this anywhere, how does TestDynamicTopicReceiver know it is "test.topic"?

      @Autowired
      private Topic destination;

      Delete
    2. Hi Ziyang,

      Thanks to you for reading it :)

      What I am doing in TestDynamicReceiver is autowiring by type. This means it will look at the Spring container for a bean of type javax.jms.Topic. If you review the config file "jms-config.xml", you will notice that there's only one qualifying bean, which is testTopic (ActiveMQTopic implements javax.jms.Topic). The other two destinations implement javax.jms.Queue.

      Delete
    3. Thanks for the explanation, as I expected, :), in fact, I never used by type in real world.

      I like your blogs very much! Hope you keep on.

      Happy New Year!

      Delete
  3. Hi i am Nagarjuna, superb work, no words to say simply superb explined perfectly.
    I have a doubt how you configured destination and where you created.
    Thank you

    ReplyDelete
    Replies
    1. Hi Nagarjuna,

      Thanks for your feedback, I appreciate it. I will try to answer your question:

      In synchronous reception you can see the destination defined as a queue named "test.sync.queue" (3.1.1 section). That's the bean named syncTestQueue. This bean is configured in the jmsTemplate as its default destination, so when you use the template to send a message, you can skip the destination and it will use the default (see the producer in 3.1.2 section). On the other hand, the consumer (section 3.1.3) receives the message from the queue passed to the jmsTemplate. The template will use its destination resolver to resolve the name passed to a JMS destination.

      In asynchronous reception, the mechanism is the same.

      Did I answer your question?

      Delete
  4. Hi Xavier,
    Thank you very much for this helpful and useful post!
    We're trying to configure our application to use JMS topics, but it seems that some of the messages get lost and don't get consumed (it doesn't happen when we use queues).
    For detailed information see my post: http://stackoverflow.com/questions/21073479/activemq-with-jms-topic-some-messages-not-dequeued-by-consumer
    Do you maybe have an idea what could be the problem? Any help would be very much appreciated. :)
    Thanks!

    ReplyDelete
    Replies
    1. Hi Ayelet,

      I've added a test called "TestIterateToTopic" to my project, but doesn't seem to reproduce your issue since all messages are being received by the consumer. I'm using a local JMS transaction. Maybe, as you noted in SO, could be something with Atomikos configuration. I'm sorry for not being more helpful.

      Please, let me know if you find what's wrong.

      Delete
    2. Hi Xavier,
      We couldn't figure out what the problem was, but when we used virtual topic instead of regular topic, it seemed to solve the issue.

      Delete
  5. Extremely helpful. Even more so than Spring's documentation and parts of the Spring in Action book.

    ReplyDelete
    Replies
    1. Thank you jackcd1, I am very pleased that it was helpful to you. I appreciate your words :)

      Delete
  6. Thanks by tutorial, very valuable and elegant. I am going to share this tutorial this weekend through twitter and SpringSource group in facebook.

    ReplyDelete
    Replies
    1. Many thanks Manuel, I appreciate you share it :)

      Delete
    2. I invite you to be a member at "https://www.facebook.com/groups/springsource/" I think is fair if you are part of the community and let us read your pretty cool tutorials! Let me know if you have a twitter account. Kind Regards.

      Delete
    3. Manuel, thank you for the invitation but, as weird as it may seem, I don't currently have Facebook! I will think about it. Regarding the twitter account, my twitter handle is @xavips. Best regards.

      Delete
    4. No problem, I am going to share your work anyway on FB, excellent material. Thanks again by your work!

      Delete