JMS and the Importance of Messaging

This chapter is from the book

Special Edition Using Java 2, Enterprise Edition

Learn More Buy

This chapter is from the book

This chapter is from the book 

Special Edition Using Java 2, Enterprise Edition

Learn More Buy

The Java Message Service (JMS)

An enterprise-level platform such as J2EE wouldn't be complete without a messaging API. Thanks to the Java Message Service (JMS), J2EE has good support for messaging. JMS is similar to JDBC in that it is a standard API for existing message systems, just as JDBC is a standard API for accessing databases. Like JDBC, JMS can't stand on its own two feet. It only defines the interfaces and major classes that are used to communicate with a messaging system, and does not actually implement any messaging.

Most of the top J2EE application servers support messaging, so you shouldn't have too hard a time finding an implementation of the messaging service. Sun provides a list of JMS vendors at http://java.sun.com/products/jms/vendors.html.

JMS supports both point-to-point messaging and publish-subscribe. You send point-to-point messages using queues and you publish messages to subscribers via topics.

JMS seems a little complicated when you first start using it because it you must create a lot of classes just to send a message. After you understand the structure of the classes, however, it is easy to use. The QueueConnection and TopicConnection classes manage all your interactions with the message server, but to interact with the message server, you must create either a QueueSession or a TopicSession. Figure 19.6 shows the relationship between connections, sessions, queues, and topics.

Figure 19.6. Although the connection is your link to the message server, you do your work using other objects.

In many ways, JMS connections are like JDBC connections. They represent your connection to a server and you use them to create other objects, but you don't normally do your work by using them directly. The QueueSession and TopicSession classes are also similar to JDBC connections in that you use them to create objects, but again, you don't interact with them directly to send and receive messages.

The session objects let you create the senders, receivers, queues, and topics you need to actually send and receive messages.

One of the big benefits of JMS is that you can perform message operations as part of a transaction. When you create a QueueSession or a TopicSession, you can specify that the session is transactional. The reason transactions are so important for messaging is that you get into situations in which you read a message off a message queue and then try to insert the message in a database. If the database operation fails, you can't just stick the message back on the queue. If the operation is transactional, however, when the database operation fails, your message stays in the queue. Transactions can also ensure that the messaging system delivers the messages in the order you send them.

Sending Queue Messages

Before you can perform any kind of queue operations in JMS, you must first create a QueueConnection. You use JNDI to locate a QueueConnectionFactory and then create the connection. Next, you use the QueueConnection to create a QueueSession. Next, you create the Queue (although you usually look for the queue in the naming service first, in case someone has already created it). When you create the queue, you must give it a name. Every queue must have a unique name.

After you have a Queue, you're almost ready. All you need is a QueueSender to send messages on the queue and one or more Message objects. You use the QueueSession to create both the QueueSender and the messages.

Listing 19.1 shows a message queue version of the ever-popular "Hello World" program. In this case, the program sends "Hello" messages over a message queue.

Listing 19.1 Source Code for HelloQueueSender.java

package usingj2ee.messages;

import javax.jms.*;
import javax.naming.*;

public class HelloQueueSender
{
    public static void main(String[] args)
    {
        try
        {
// Locate the JNDI naming service
            Context ctx = new InitialContext();

// Locate the Queue Connection Factory via JNDI
            QueueConnectionFactory factory =
                (QueueConnectionFactory) ctx.lookup(
                    "javax.jms.QueueConnectionFactory");

// Create a new Queue Connection
            QueueConnection conn = factory.createQueueConnection();

// Create a Queue Session, ask JMS to acknowledge the messages
// The session is non-transactional; you don't send messages
// as part of a transaction.
            QueueSession session = conn.createQueueSession(false,
                Session.AUTO_ACKNOWLEDGE);

            Queue queue = null;

            try
            {
// See if someone has already created the queue
                queue = (Queue) ctx.lookup("HelloQueue");
            }
            catch (NameNotFoundException exc)
            {

// If not, create a new Queue and store it in the JNDI directory
                queue = session.createQueue("HelloQueue");
                ctx.bind("HelloQueue", queue);
            }

// Create a simple text message
            TextMessage message = session.createTextMessage("Hello Client!");

// Create a QueueSender (so you can send messages)
            QueueSender sender = session.createSender(queue);

// Tell the Queue Connection you are ready to interact with the message service
            conn.start();

            for (;;)
            {
// Send a message
                sender.send(message);

// Wait 5 seconds (5,000 milliseconds) before sending another message
                try { Thread.sleep(5000); } catch (Exception ignore) {}
            }
        }
        catch (Exception exc)
        {
            exc.printStackTrace();
        }
    }
}
						

Receiving Queue Messages

Most of the setup you do for sending messages is the same as for receiving them. Of course, instead of creating a QueueSender, you create a QueueReceiver. There are two different ways to receive messages. You can call the receive method, which waits for messages, or you can create a listener object that receives messages when they become available.

Listing 19.2 shows a simple message receiver that uses the receive method to retrieve the queue messages. Notice that most of the setup is the same as in Listing 19.1.

Listing 19.2 Source Code for HelloQueueReceiver.java

package usingj2ee.messages;

import javax.jms.*;
import javax.naming.*;

public class HelloQueueReceiver
{
    public static void main(String[] args)
    {
        try
        {
// Locate the JNDI naming service
            Context ctx = new InitialContext();

// Locate the Queue Connection Factory via JNDI
            QueueConnectionFactory factory =
                (QueueConnectionFactory) ctx.lookup(
                    "javax.jms.QueueConnectionFactory");

// Create a new Queue Connection
            QueueConnection conn = factory.createQueueConnection();

// Create a Queue Session, ask JMS to acknowledge the messages
// This program receives messages, so it doesn't really care.
// The session is non-transactional 

            QueueSession session = conn.createQueueSession(false,
                Session.AUTO_ACKNOWLEDGE);

            Queue queue = null;
            try
            {
// See if someone has already created the queue
                queue = (Queue) ctx.lookup("HelloQueue");
            }
            catch (NameNotFoundException exc)
            {
// If not, create a new Queue and store it in the JNDI directory
                queue = session.createQueue("HelloQueue");
                ctx.bind("HelloQueue", queue);
            }

// Create a QueueReceiver to receive messages
            QueueReceiver receiver = session.createReceiver(queue);

// Tell the Queue Connection you are ready to interact with the message service
            conn.start();

            for (;;)
            {
// Receive the next message
                TextMessage message = (TextMessage) receiver.receive();

// Print the message contents
                System.out.println(message.getText());
            }
        }
        catch (Exception exc)
        {
            exc.printStackTrace();
        }
    }
}
						

If you don't want to wait for messages, but instead prefer to have the QueueReceiver notify you when a message comes in, you can implement the MessageListener interface. You tell the QueueReceiver about your message listener and it will automatically let you know when a message comes in. Listing 19.3 shows the listener version of the message receiver in Listing 19.2.

Listing 19.3 Source Code for HelloQueueListener.java

package usingj2ee.messages;

import javax.jms.*;
import javax.naming.*;

public class HelloQueueListener implements MessageListener
{
    HelloQueueListener()
    {
    }

/** Called by the QueueReceiver to handle the next message in the queue */
    public void onMessage(Message message)
    {
        try
        {
// Assume the message is a text message
            TextMessage textMsg = (TextMessage) message;

// Print the message text
            System.out.println(textMsg.getText());
        }
        catch (JMSException exc)
        {
            exc.printStackTrace();
        }
    }

    public static void main(String[] args)
    {
        try
        {
// Locate the JNDI naming service
            Context ctx = new InitialContext();

// Locate the Queue Connection Factory via JNDI
            QueueConnectionFactory factory =
                (QueueConnectionFactory) ctx.lookup(
                    "javax.jms.QueueConnectionFactory");

// Create a new Queue Connection
            QueueConnection conn = factory.createQueueConnection();

// Create a non-transactional Queue Session
            QueueSession session = conn.createQueueSession(false,
                Session.AUTO_ACKNOWLEDGE);

            Queue queue = null;
            try
            {
// See if someone has already created the queue
                queue = (Queue) ctx.lookup("HelloQueue");
            }
            catch (NameNotFoundException exc)
            {
// If not, create a new Queue and store it in the JNDI directory
                queue = session.createQueue("HelloQueue");
                ctx.bind("HelloQueue", queue);
            }

// Create a QueueReceiver to receive messages
            QueueReceiver receiver = session.createReceiver(queue);

// Tell the Queue Connection you are ready to interact with the message service
            conn.start();


// Tell the receiver to call your listener object when a new message arrives
            receiver.setMessageListener(new HelloQueueListener());

// Normally you would do other processing here...
            Thread.sleep(999999999);
        }
        catch (Exception exc)
        {
            exc.printStackTrace();
        }
    }
}
						

You typically have one queue receiver and one or more queue senders. Although JMS allows you to have multiple receivers on a queue, it is up to the individual implementations to decide how to handle multiple receivers. Some might distribute the messages evenly across all receivers, and others might just send all the messages to the first receiver you create.

Publishing Messages

Although the overall concept of publish-subscribe is a bit different from point-to-point messages, the JMS calls for creating a topic and publishing messages are remarkably similar to the calls for sending queue messages. In fact, if you go through the program in Listing 19.1 earlier in this chapter and change all the occurrences of Queue to Topic, you'll almost have a working topic publisher. You must also change the createSender call to createPublisher.

As with queues, each topic must have a unique name. Unlike queues, in which you only have one receiver, you normally have many subscribers to a topic. You can also have multiple publishers.

Listing 19.4 shows the publish-subscribe version of the ubiquitous "Hello World," at least the publishing half.

Listing 19.4 Source Code for HelloTopicPublisher.java

package usingj2ee.messages;

import javax.jms.*;
import javax.naming.*;

public class HelloTopicPublisher
{
    public static void main(String[] args)
    {
        try
        {
// Locate the JNDI naming service
            Context ctx = new InitialContext();

// Locate the Topic Connection Factory via JNDI
            TopicConnectionFactory factory =
                (TopicConnectionFactory) ctx.lookup(
                    "javax.jms.TopicConnectionFactory");

// Create a new Topic Connection
            TopicConnection conn = factory.createTopicConnection();

// Create a non-transactional TopicSession
            TopicSession session = conn.createTopicSession(false,
                Session.AUTO_ACKNOWLEDGE);

            Topic topic = null;

            try
            {
// See if someone has already created the topic
                topic = (Topic) ctx.lookup("HelloTopic");
            }
            catch (NameNotFoundException exc)
            {
// If not, create a new topic and store it in the JNDI directory
                topic = session.createTopic("HelloTopic");
                ctx.bind("HelloTopic", topic);
            }

// Create a new text message
            TextMessage message = session.createTextMessage("Hello Client!");

// Create a publisher to publish messages
            TopicPublisher sender = session.createPublisher(topic);

// Tell the Topic Connection you are ready to interact with the message service
            conn.start();

            for (;;)
            {
// Publish the message
                sender.publish(message);

// Wait 5 seconds before sending another message
                try { Thread.sleep(5000); } catch (Exception ignore) {}
            }
        }
        catch (Exception exc)
        {
            exc.printStackTrace();
        }
    }
}
						

You can run multiple copies of this program with no problems. Your clients will just see more messages than they do if only one copy is running.

Subscribing to Topics

Subscribing to a topic is just as easy as listening on a queue. Again, you can just about convert the program in Listing 19.2 earlier in the chapter to work with topics just by changing all the occurrences of Queue to Topic. Listing 19.5 shows the subscriber for the "Hello World" pub-sub example.

Listing 19.5 Source Code for HelloTopicReceiver.java

package usingj2ee.messages;

import javax.jms.*;
import javax.naming.*;

public class HelloTopicReceiver
{
    public static void main(String[] args)
    {
        try
        {
// Locate the JNDI naming service
            Context ctx = new InitialContext();

// Locate the Topic Connection Factory via JNDI
            TopicConnectionFactory factory =
                (TopicConnectionFactory) ctx.lookup(
                    "javax.jms.TopicConnectionFactory");

// Create a new Topic Connection
            TopicConnection conn = factory.createTopicConnection();

// Create a non-transactional Topic Session
            TopicSession session = conn.createTopicSession(false,
                Session.AUTO_ACKNOWLEDGE); 

            Topic topic = null;
            try
            {
// See if someone has already created the topic
                topic = (Topic) ctx.lookup("HelloTopic");
            }
            catch (NameNotFoundException exc)
            {
// If not, create a new topic and store it in the JNDI directory
                topic = session.createTopic("HelloTopic");
                ctx.bind("HelloTopic", topic);
            }

// Create a subscriber to receive messages
            TopicSubscriber subscriber = session.createSubscriber(topic);

// Tell the Topic Connection you are ready to interact with the message service
            conn.start();

            for (;;)
            {
// Get the next published message
                TextMessage message = (TextMessage) subscriber.receive();

// Print the message text
                System.out.println(message.getText());
            }
        }
        catch (Exception exc)
        {
            exc.printStackTrace();
        }
    }
}
						

As with the QueueReceiver class, you can choose to receive message notifications asynchronously by using the MessageListener interface. The actual listener implementation for topics is identical to the one for queues.

Durable Subscriptions

Most pub-sub systems you see today deal with frequently published data and tend to provide real-time monitoring and status updates. Pub-sub is good for these kinds of operations. Many pub-sub implementations have a peculiar limitation that makes them only suitable for applications with frequently published data-more specifically, applications in which it doesn't matter if you miss a message because there will be another one in a few minutes.

The reason pub-sub usually only works in these types of applications is that pub-sub doesn't usually have the concept of a persistent, or durable subscription. For example, suppose you are publishing flight cancellations for LaGuardia Airport. Although it's true that they do seem to occur often enough that there will be another one in a few minutes, you really don't want to miss one. Now, suppose you have a program that makes a list of passengers who need to be rebooked for the next available flight (tomorrow, next week, sometime next year, and so on). If the program shuts down for a few minutes, it might miss three or four cancel lations! You want the message server to hold on to your messages while your program is down.

JMS supports durable subscription. After you create a durable subscription, the JMS server keeps any messages you miss while your program isn't running. To create a durable subscription, use the createDurableSubscriber method in the TopicSession class:

						
public TopicSubscriber createDurableSubscriber(Topic topic,
    String subscriptionName)

You must always use the same subscription name when reconnecting your program to its durable subscription. That is, if you call the subscription RebookFlightCx (that's the subscription name, not the topic name), you must always ask for RebookFlightCx when resuming that subscription. Keep in mind that durable connections are expensive for the server to maintain, so only use them when absolutely necessary.

Messages can have an expiration time, so when a message expires, the server can safely remove it from the durable subscription. This helps keep the database size down when you don't resume a durable subscription for a long time.