Message queues
- Buy JMS Books from Amazon Book Store
Like we mentioned earlier, message queues are used when our JMS code uses the
point-to-point (PTP) messaging domain. For the PTP messaging domain, there is
usually one message producer and one message consumer. The message producer
and the message consumer don’t need to run concurrently in order to communicate.
The messages placed in the message queue by the message producer will stay in the
message queue until the message consumer executes and requests the messages from
the queue.
Sending messages to a message queue
The following example illustrates how to add messages to a mess age queue:
package net.ensode.glassfishbook;
import javax.annotation.Resource;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
public class MessageSender
{
@Resource(mappedName = “jms/GlassFishBookConnectionFactory”)
private static ConnectionFactory connectionFactory;
@Resource(mappedName = “jms/GlassFishBookQueue”)
private static Queue queue;
public void produceMessages()
{
MessageProducer messageProducer;
TextMessage textMessage;
try
{
Connection connection = connectionFactory.createConnection();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
messageProducer = session.createProducer(queue);
textMessage = session.createTextMessage();
textMessage.setText(“Testing, 1, 2, 3. Can you hear me?”);
System.out.println(“Sending the following message: ”
+ textMessage.getText());
messageProducer.send(textMessage);
textMessage.setText(“Do you copy?”);
System.out.println(“Sending the following message: ”
+ textMessage.getText());
messageProducer.send(textMessage);
textMessage.setText(“Good bye!”);
System.out.println(“Sending the following message: ”
+ textMessage.getText());
messageProducer.send(textMessage);
messageProducer.close();
session.close();
connection.close();
}
catch (JMSException e)
{
e.printStackTrace();
}
}
public static void main(String[] args)
{
new MessageSender().produceMessages();
}
}
Before delving into the details of this code, alert readers might have noticed that this
class is a standalone Java application as it contains a main method. As this class is
standalone, it executes outside the application server. In spite of this, we can see that
some resources are injected into it, specifically the connection factory and queue.
The reason we can inject resources into this code, even though it runs outside the
application server, is because GlashFish includes a utility called appclient.
This utility allows us to “wrap” an executable JAR file and allows it to have access
to the application server resources. To execute the previous code, assuming it is
packaged in an executable JAR file called jmsptpproducer.jar, we would type
the following command in the command line:
appclient -client jmsptpproducer.jar
We would then see, after some GlashFish log entries, the following output on
the console:
Sending the following message: Testing, 1, 2, 3. Can you hear me?
Sending the following message: Do you copy?
Sending the following message: Good bye!
The appclient executable can be found under [GlashFish installation
directory]/GlashFish/bin. The previous example assumes this directory is in
your PATH variable. If it isn’t the complete path to the appclient executable, it must
be typed in the command line.
With that out of the way, we can now explain the code.
The produceMessages() method performs all the necessary steps to send messages
to a message queue.
The first thing this method does is obtain a JMS connection by invoking
the createConnection() method on the injected instance of javax.jms.
ConnectionFactory. Notice that the mappedName attribute of the @Resource
annotation decorating the connection factory object matches the JNDI name of
the connection factory we set up in the GlashFish web console. Behind the scenes,
a JNDI lookup is made using this name to obtain the connection factory object.
After obtaining a connection, the next step is to obtain a JMS session from said
connection. This can be accomplished by calling the createSession() method on
the Connection object. As can be seen in the previous code, the createSession()
method takes two parameters.
The first parameter of the createSession() method is a Boolean indicating if the
session is transacted. If this value is true, several messages can be sent as part of a
transaction by invoking the commit() method in the session object. Similarly, they
can be rolled back by invoking its rollback() method.
The second parameter of the createSession() method indicates how messages are
acknowledged by the message receiver. Valid values for this parameter are defined
as constants in the javax.jms.Session interface.
- Session.AUTO_ACKNOWLEDGE: indicates that the session will automatically
acknowledge the receipt of a message. - Session.CLIENT_ACKNOWLEDGE: indicates that the message receiver must
explicitly call the acknowledge() method on the message. - Session.DUPS_OK_ACKNOWLEDGE: indicates that the session will lazily
acknowledge the receipt of messages. Using this value might result in some
messages being delivered more than once.
After obtaining a JMS session, an instance of javax.jms.MessageProducer is
obtained by invoking the createProducer() method on the session object. The
MessageProducer object is the one that will actually send messages to the message
queue. The injected Queue instance is passed as a parameter to the createProducer()
method . Again, the value of the mappedName attribute for the @Resource annotation
decorating this object must match the JNDI name we gave our message queue when
setting it up in the GlashFish web console.
After obtaining an instance of MessageProducer, the code creates a series of text
messages by invoking the createTextMessage() method on the session object. This
method returns an instance of a class implementing the javax.jms.TextMessage
interface. This interface defines a method called setText() , which is used to set the
actual text in the message. After creating each text message and setting its text, they are
sent to the queue by invoking the send() method on the MessageProducer object.
After sending the messages, the code disconnects from the JMS queue by invoking
the close() method on the MessageProducer object , on the Session object , and on
the Connection object.
Although the previous example sends only text messages to the queue, we are not
limited to this type of message. The JMS API provides several types of messages
that can be sent and received by JMS applications. All message types are defined as
interfaces in the javax.jms package.
The following table lists all the available message types:
For more information on all of these message types, consult their JavaDoc
documentation at http://java.sun.com/javaee/6/docs/api/.
Retrieving messages from a message queue
There is no point in sending messages from a queue if nothing is going to receive
them. The following example illustrates how to retrieve messages from a JMS
message queue:
package net.ensode.glassfishbook;
import javax.annotation.Resource;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
public class MessageReceiver
{
@Resource(mappedName = “jms/GlassFishBookConnectionFactory”)
private static ConnectionFactory connectionFactory;
@Resource(mappedName = “jms/GlassFishBookQueue”)
private static Queue queue;
public void getMessages()
{
Connection connection;
MessageConsumer messageConsumer;
TextMessage textMessage;
boolean goodByeReceived = false;
try
{
connection = connectionFactory.createConnection();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
messageConsumer = session.createConsumer(queue);
connection.start();
while (!goodByeReceived)
{
System.out.println(“Waiting for messages…”);
textMessage = (TextMessage) messageConsumer.receive();
if (textMessage != null)
{
System.out.print(“Received the following message: “);
System.out.println(textMessage.getText());
System.out.println();
}
if (textMessage.getText() != null
&& textMessage.getText().equals(“Good bye!”))
{
goodByeReceived = true;
}
}
messageConsumer.close();
session.close();
connection.close();
}
catch (JMSException e)
{
e.printStackTrace();
}
}
public static void main(String[] args)
{
new MessageReceiver().getMessages();
}
}
Just like in the previous example, an instance of javax.jms.ConnectionFactory
and an instance of javax.jms.Queue are injected by using the @Resource
annotation. Getting a connection and a JMS session is exactly the same as in
the previous example.
In this example, we obtain an instance of javax.jms.MessageConsumer by calling
the createConsumer() method on the JMS session object. When we are ready to
start receiving messages from the message queue, we need to invoke the start()
method on the JMS connection object.
Code not receiving messages?
A common mistake when writing JMS messages is to fail to call the
start() method on the JMS connection object. If our code is not
receiving messages it should be receiving, we need to make sure we
didn’t forget to call this method.
Messages are received by invoking the receive() method on the instance of
MessageConsumer obtained from the JMS session. This method returns an instance
of a class implementing the javax.jms.Message interface. It must be casted to the
appropriate type in order to obtain the actual message.
In this particular example, we placed this method call in a while loop, as we are
expecting a message that will let us know that no more messages are coming.
Specifically, we are looking for a message containing the text “Good bye!”. Once
we receive said message, we break out of the loop and continue processing. In this
particular case, there is no more processing to do. Therefore, all we do is call the
close() method on the message consumer object, on the session object, and on the
connection object.
Just like in the previous example, using the appclient utility allows us to inject
resources into the code and prevents us from having to add any libraries to the
CLASSPATH. After executing the code through the appclient utility, we should
see the following output in the command line:
appclient -client target/jmsptpconsumer.jar
Waiting for messages…
Received the following message: Testing, 1, 2, 3. Can you hear me?
Waiting for messages…
Received the following message: Do you copy?
Waiting for messages…
Received the following message: Good bye!
This of course assumes that the previous example was already executed and it placed
the messages in the message queue.
Asynchronously receiving messages from a message queue
The MessageConsumer.receive() method has a disadvantage—it blocks
execution until a message is received from the queue. We can avoid this
disadvantage by receiving messages asynchronously via an implementation
of the javax.jms.MessageListener interface.
The javax.jms.MessageListener interface contains a single method called
onMessage. It takes an instance of a class implementing the javax.jms.Message
interface as its sole parameter. The following example illustrates a typical
implementation of this interface:
package net.ensode.glassfishbook;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class ExampleMessageListener implements MessageListener
{
@Override
public void onMessage(Message message)
{
TextMessage textMessage = (TextMessage) message;
try
{
System.out.print(“Received the following message: “);
System.out.println(textMessage.getText());
System.out.println();
}
catch (JMSException e)
{
e.printStackTrace();
}
}
}
In this case, the onMessage() method simply outputs the message text to the console.
Our main code can now delegate message retrieval to our custom MessageListener
implementation:
package net.ensode.glassfishbook;
import javax.annotation.Resource;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
public class AsynchMessReceiver
{
@Resource(mappedName = “jms/GlassFishBookConnectionFactory”)
private static ConnectionFactory connectionFactory;
@Resource(mappedName = “jms/GlassFishBookQueue”)
private static Queue queue;
public void getMessages()
{
Connection connection;
MessageConsumer messageConsumer;
try
{
connection = connectionFactory.createConnection();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
messageConsumer = session.createConsumer(queue);
messageConsumer.setMessageListener(new
ExampleMessageListener());
connection.start();
System.out.println(“The above line wil l allow the ”
+ “MessageListener implementation to ”
+ “receiving and processing messages from the queue.”);
Thread.sleep(1000);
System.out.println(“Our code does not have to block ”
+ “while messages are received.”);
Thread.sleep(1000);
System.out.println(“It can do other stuff ”
+ “(hopefully something more useful than sending ”
+ “silly output to the console. 
”);
Thread.sleep(1000);
messageConsumer.close();
session.close();
connection.close();
}
catch (JMSException e)
{
e.printStackTrace();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
public static void main(String[] args)
{
new AsynchMessReceiver().getMessages();
}
}
The only relevant difference between this example and the one in the previous
section is that in this case, we are calling the setMessageListener() method on
the instance of javax.jms.MessageConsumer obtained from the JMS session. We
pass an instance of our custom implementation of javax.jms.MessageListener
to this method. Its onMessage() method is automatically called whenever there is a
message waiting in the queue. By using this approach, the main code does not block
while waiting to receive messages.
Executing the previous example (using of course GlashFish‘s appclient utility)
results in the following output:
appclient -client target/jmsptpasynchconsumer.jar
The above line will allow the MessageListener implementation to receiving
and processing messages from the queue.
Received the following message: Testing, 1, 2, 3. Can you hear me?
Received the following message: Do you copy?
Received the following message: Good bye!
Our code does not have to block while messages are received.
It can do other stuff (hopefully something more useful than sending silly
output to the console.
Notice how the messages were received and processed while the main thread was
executing. We can tell this is the case because the output of the onMessage() method
of our MessageListener can be seen between calls to System.out.println() in the
primary class.
Browsing message queues
JMS provides a way to browse message queues without actually removing the
messages from the queue. The following example illustrates how to do this:
package net.ensode.glassfishbook;
import java.util.Enumeration;
import javax.annotation.Resource;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TextMessage;
public class MessageQueueBrowser
{
@Resource(mappedName = “jms/GlassFishBookConnectionFactory”)
private static ConnectionFactory connectionFactory;
@Resource(mappedName = “jms/GlassFishBookQueue”)
private static Queue queue;
public void browseMessages()
{
try
{
Enumeration messageEnumeration;
TextMessage textMessage;
Connection connection = connectionFactory.createConnection();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
QueueBrowser browser = session.createBrowser(queue);
messageEnumeration = browser.getEnumeration();
if (messageEnumeration != null)
{
if (!messageEnumeration.hasMoreElements())
{
System.out.println(“There are no messages ” + “in the
queue.”);
}
else
{
System.out.println(“The following messages are in the
queue:”);
while (messageEnumeration.hasMoreElements())
{
textMessage =
(TextMessage) messageEnumeration.nextElement();
System.out.println(textMessage.getText());
}
}
}
session.close();
connection.close();
}
catch (JMSException e)
{
e.printStackTrace();
}
}
public static void main(String[] args)
{
new MessageQueueBrowser().browseMessages();
}
}
As we can see, the procedure to browse messages in a message queue is
straightforward. We obtain a JMS connection and a JMS session the usual way, then
invoke the createBrowser() method on the JMS session object. This method returns
an implementation of the javax.jms.QueueBrowser interface. This interface contains
a getEnumeration() method that we can invoke to obtain an enumeration containing
all messages in the queue. To examine the messages in the queue, we simply traverse
this enumeration and obtain the messages one by one. In the previous example, we
simply invoke the getText() method of each message in the queue.
JMS Articles
- Integrating Spring with JMS by Christy
- JMS Articles
- Spring JMS API and Weblogic JMS Integration by Sangeetha
September 10, 2010
JMS