Table of Contents

Introduction
JMS Architecture
    JMS Messaging Models
JMS Example
    JMS Provider: JBoss configuration
    Core JMS API
    Client: JMS Producer
    Client: JMSConsumer
    JMS Unified Client API
    Execute the example
    Client: JMSConsumer revisited
    JMS Topics instead of Queues
Reliability - Durable Subscriptions
Message Driven Beans
    Client: JMS Consumer revisited
    Transactions
    Distributed Transactions
Summary
References

Introduction

JMS or Java Messaging System is an asynchronous communication API for Java applications. The JMS Specification is defined in JSR 914 [1]. The Java Message Service was developed by Sun Microsystems to provide a means for Java programs to access enterprise messaging systems or message oriented middleware (MOM). They provide a mechanism for integrating applications in a loosely coupled way. They provide asynchronous delivery of data between applications on a store and forward basis; i.e., the applications do not communicate directly with each other, but instead communicate with the MOM, which acts as an intermediary [2], thus handling all network communication details for you. If e.g. the network connection is not available, the MOM will store the message until the connection becomes available, and then forward it to the destination. Thus, the JMS API is characterised as asynhronous and reliable (it can ensure that a message is delivered once and only once).
Even though Messaging follows the same principles as electronic mail (email), messaging differs from electronic mail, which is a method of communication between people or between software applications and people. Messaging is used for communication between software applications or software components.

In the following we 'll show a simple Java "Hello JMS" message exchange application using JMS 1.1.

JMS Architecture

• JMS Provider: A messaging system that implements the JMS specification. A JMS Provider, also known as a JMS Server, routes messages between clients.
• Clients: Java applications that send or receive JMS messages. A message sender is called the Producer, and the recipient is called a Consumer.
• Messages: Messages contain data or events exchanged between Producers and Consumers.
• Destinations: A Producer sends a message to a JMS Destination (either a Queue or a Topic), and the Consumer(s) listening on the JMS Destination receives the message.

JMS Messaging Models

JMS has two messaging models [3]: Point-to-Point (P2P) and Publish-Subscribe (Pub-Sub). P2P is a traditional one-to-one queueing mechanism, i.e. although multiple Consumers can listen on a queue, only one Consumer receives a particular message. Producers send messages to a queue, and the JMS Provider (aka JMS Server) delivers each message sequentially to a Consumer listening on the queue. The following figure shows the relationships between Point-to-Point Producers and Consumers.

Figure 1 JMS Queue

Publish-Subscribe is a one-to-many broadcast model, similar to a newsgroup or a bulletin board or an RSS newsfeed. Producers publish messages to a topic, and the JMS Server delivers messages sequentially to those Consumers subscribed to that topic. The following figure shows the relationships between Pub-Sub Producers and Consumers.

Figure 2 JMS Topic

The biggest difference between P2P and Pub-Sub is that in the Publish-Subscribe Model, all Consumers that subscribe to a Topic can receive all messages published to that topic, while with Point-to-Point, only one Consumer on a queue receives a particular message.

Even though JMS is inherently asynchronous, the JMS specification allows for messages to be consumed in either of two ways [10]:

• Synchronously: A subscriber or a receiver explicitly fetches the message from the destination by calling the receive method. The receive method can block until a message arrives or can time out if a message does not arrive within a specified time limit.

• Asynchronously: A client can register a message listener with a consumer. A message listener is similar to an event listener. Whenever a message arrives at the destination, the JMS provider delivers the message by calling the listener's onMessage method, which acts on the contents of the message.

• An object must implement java.io.Serializable.
• Each data member must be serializable. By default, String, the Java primitives (int, float, and so on) and the Java primitive wrappers (Integer, Float, and so on) are all serializable.

JMS Example

• JMS Provider: The JMS Provider will be JBoss, a full J2EE application server which provides support for JMS 1.1.
• Clients: Our clients will be a JMSProducer and a JMSConsumer java class. 
• Messages: The JMSProducer will send a simple "Hello JMS" TextMessage to be delivered to the JMSConsumer.
• Destinations: The JMS Destination will be a queue, named MyQueueXA. The JMSProducer will post to this queue and the JMSConsumer will listen to this queue to consume the message.

JMS Provider: JBoss configuration

To execute and test the program, you need access to a vendor implementation of JMS. Most Java 2 Enterprise Edition (J2EE) application servers provide an implementation of JMS. JBoss is such an application server. JBoss Messaging provided by default with JBoss 4.2 supports the latest JMS 1.1 specification and it offers vastly improved performance in both single node and clustered environments compared to the older JBossMQ JBoss JMS provider [9]. However, this is something that doesn't bother us as users of the JBoss JMS service.

To setup a JMS message queue or topic in JBoss is very easy. There are actually 3 ways to do this [3,7,8,9]:

1. Using the JMX console: Start the JBoss application server (%JBOSS_HOME%/bin/run.bat or $JBOSS_HOME/bin.run.sh) and open the JMX console (http://localhost:8080/jmx-console). Click on the service=DestinationManager link (under the jboss.mq heading) to see the JBossMQ DestinationManager MBean page. Look for the createQueue( ) operation and enter jms/MyQueue for the first parameter (the J2EE JNDI name would then be java:comp/env/jms/MyQueue), and queue/MyQueue for the second parameter (the JBoss-specific JNDI name). The JMX-console has the advantage that it is easy to use and you can dynamically specify JMS resources; the disadvantage though is that your changes are lost once you shut down JBoss
2. Add an <mbean> element for the queue to $JBOSS_HOME/server/default/deploy/jms/jbossmq-destinations-service.xml file. This file is part of the core JBoss deployment and contains JBoss-specific default test Queues and Topics. This has the advantage that the changes survive JBoss startup/shutdowns; the disandvantage is however that you mix-up your destinations with JBoss internal destinations.

<mbean code="org.jboss.mq.server.jmx.Queue" name="jboss.mq.destination:service=Queue,name=MyQueue">   <depends optional-attribute-name="DestinationManager"> jboss.mq:service=DestinationManager</depends> </mbean>

3. The best solution is to create your own service descriptor file (post-fixed with -service.xml) that resides in $JBOSS_HOME/server/default/deploy and add an <mbean> element for the queue. This way your changes survice JBoss startup/shutdowns and  you don't mix-up your destinations with JBoss internal destinations. Co-mingling destinations is bad because each time you upgrade to a new version of JBoss, you have to re-add the <mbean> elements to your service descriptor to make things work again.

   <?xml version="1.0" encoding="UTF-8"?>
   <server>
      <mbean code="org.jboss.mq.server.jmx.Queue" name="jboss.mq.destination:service=Queue,name=MyQueue">
        <depends optional-attribute-name="DestinationManager"> jboss.mq:service=DestinationManager</depends>
      </mbean>
   </server> 

That's it! Please note that destination without a configured SecurityManager or without a SecurityConf will default to role guest with read=true, write=true, create=false. If you want to setup a SecurityManager, you should modify your mbean similar to this:

<?xml version="1.0" encoding="UTF-8"?>
<server>
   <mbean code="org.jboss.mq.server.jmx.Queue" name="jboss.mq.destination:service=Queue,name=MyQueue">
     <depends optional-attribute-name="DestinationManager"> jboss.mq:service=DestinationManager</depends>
     <depends optional-attribute-name="SecurityManager">jboss.mq:service=SecurityManager</depends>
        <attribute name="SecurityConf">
            <security>
                <role name="guest" create="false" read="true" write="true"/>
                <role name="senders" create="false" read="false" write="true"/>
                <role name="receivers" create="false" read="true" write="false"/>
            </security>
        </attribute>
   </mbean>
</server> 

Figure 3 HSQL DB Manager

As you can see, JBoss uses 5 tables for JMS handling, which are shown in the following ER diagram.

Figure 4 JMS Entity-Relationship model used by JBoss

Core JMS API

• Message: Holds business data and routing information. Although you'll find several types of Messages, most of the time you'll use either a TextMessage (that contains textual data in its body) or an ObjectMessage (that holds serializable objects in its body).
• Destination: Holds messages sent by the Producer to be received by the Consumer(s). A Destination is either a Queue or a Topic that the JMS Server manages on behalf of its clients.
• Connection: Enables a JMS client to send or receive Messages. Use a Connection to create one or more Sessions.
• ConnectionFactory: A ConnectionFactory is either a QueueConnectionFactory or a TopicConnectionFactory, depending on the messaging model, and it exists to create Connections.
• Session: A Session creates Producers, Consumers, and Messages. A Publish-Subscribe application uses a TopicSession, and a Point-to-Point application uses a QueueSession. Sessions are single-threaded.

Figure 5 JMS API - class diagram

Figure 6 JMS State Transition Diagram

Client: JMSProducer

1. Look up a ConnectionFactory using JNDI.
2. Get a Connection from the ConnectionFactory.
3. Create a Session associated with the Connection.
4. Look up a Destination using JNDI.
5. Create a Message Producer tied to the Destination.
6. Create a Message.
7. Send the Message.
8. Tear down the Message Producer, Session, and Connection.

package jms.producer;

import javax.jms.*;

import jms.ServiceLocator;
import jms.ServiceLocatorException;

/**
 * <code>JmsProducer</code> encapsulates sending a JMS Message.
 *
 */
public class JmsProducer {
	private static final String CONNECTION_FACTORY ="ConnectionFactory";
	private static final String MY_QUEUE = "queue/MyQueue";
    	private static final String HELLO_MSG = "Hello JMS";

	public static void main(String args[]) {
		JmsProducer.sendMessage(HELLO_MSG, CONNECTION_FACTORY, MY_QUEUE);
	}
	
	/**
	 * Making the default (no arg) constructor private
	 * ensures that this class cannnot be instantiated.
	 */
	private JmsProducer() {
	}
	/**
	 * 
	 * @param txtMessage The message (xml file) to send via JMS
	 * @param connectionFactoryJndiName connection factory JNDI name
	 * @param destinationJndiName destination JNDI name
	 * @throws JmsProducerException
	 */
	public static void sendMessage(String txtMessage,
			String connectionFactoryJndiName, String destinationJndiName)
			throws JmsProducerException {
		Connection connection = null;
		try {
			ConnectionFactory connectionFactory = null;
			Session session = null;
			Destination destination = null;
			MessageProducer messageProducer = null;
			TextMessage message = null;

			// 1. Look up a ConnectionFactory using JNDI.
			connectionFactory = ServiceLocator
					.getJmsConnectionFactory(connectionFactoryJndiName);
			
			// 2. Get a Connection from the ConnectionFactory with default user id.
			connection = connectionFactory.createConnection();
			
			// 3. Create a Session associated with the Connection.
			// 1st parameter: true if Session is transacted
			// 2nd parameter: Acknowledgement method
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
			// 4. Look up a Destination using JNDI.
			destination = ServiceLocator.getJmsDestination(destinationJndiName);
			
			// 5. Create a Message Producer tied to the Destination.
			messageProducer = session.createProducer(destination);

			// 6. Create a Message.
			message = session.createTextMessage(txtMessage);
			
			// 7. Send a Message.
			messageProducer.send(message);
		} catch (JMSException je) {
			throw new JmsProducerException(je);
		} catch (ServiceLocatorException sle) {
			throw new JmsProducerException(sle);
		} finally { // 8. Tear down the Message Producer, Session, and Connection.
			if (connection != null) {
				try { // automatically session and producer are closed
					connection.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}
}

• The first parameter is a boolean that determines whether the Session is transacted. We've set this value to false because we don't want to manage the transaction programmatically. We use Container-Managed Transactions so that JBoss manages the transaction for us.
• The second parameter sets the acknowledgment mode. Using AUTO_ACKNOWLEDGE means that once the Consumer receives a message, an acknowledgment is sent to the JMS server automatically. Usually, you should use AUTO_ACKNOWLEDGE and let the container do all the work for you.

Client: JMSConsumer

package jms.consumer;

import javax.jms.*;

import jms.ServiceLocator;
import jms.ServiceLocatorException;

/**
 * <code>JmsConsumer</code> encapsulates receiving a JMS Message.
 *
 */
public class JmsConsumer {
	private static final String QUEUE_CONNECTION_FACTORY ="ConnectionFactory";
	private static final String MY_QUEUE = "queue/MyQueue";

	public static void main(String args[]) throws Exception {
		JmsConsumer.receiveMessage(QUEUE_CONNECTION_FACTORY, MY_QUEUE);
	}
	
	/**
	 * Making the default (no arg) constructor private
	 * ensures that this class cannnot be instantiated.
	 */
	private JmsConsumer() {
	}
	/**
	 * Receives a JMS message.
	 * @param connectionFactoryJndiName connection factory JNDI name
	 * @param destinationJndiName destination JNDI name
	 * @throws JmsProducerConsumerException
	 */
	public static void receiveMessage(
			String connectionFactoryJndiName, String destinationJndiName)
			throws JmsConsumerException {
		Connection connection = null;		
		try {
			ConnectionFactory connectionFactory = null;
			Session session = null;
			Destination destination = null;
			MessageConsumer messageConsumer = null;
			TextMessage textMessage = null;

			// 1. Look up a ConnectionFactory using JNDI.
			connectionFactory = ServiceLocator
					.getJmsConnectionFactory(connectionFactoryJndiName);
			
			// 2. Get a Connection from the ConnectionFactory with default user id.
			connection = connectionFactory.createConnection();
			
			// 3. Create a Session associated with the Connection.
			// 1st parameter: true if Session is transacted
			// 2nd parameter: Acknowledgement method
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
			// 4. Look up a Destination using JNDI.
			destination = ServiceLocator.getJmsDestination(destinationJndiName);
			
			// 5. Create a Message Consumer tied to the Destination.
			messageConsumer = session.createConsumer(destination);
	
			// 6. Receive a Message.
			connection.start();
			Message message = messageConsumer.receive();
			if (message instanceof TextMessage) {
				textMessage = (TextMessage) message;
				System.out.println("Message received: " + textMessage.getText());
			}
		} catch (JMSException je) {
			throw new JmsConsumerException(je);
		} catch (ServiceLocatorException sle) {
			throw new JmsConsumerException(sle);
		} finally { // 7. Tear down the Message Consumer, Session, and Connection.
			if (connection != null) {
				try { 
//					messageProducer.close();
//					session.close();
					connection.close(); // automatically session and producer are closed
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}
}

The receive() method can be used in several ways to perform a synchronous receive. If you specify no arguments or an argument of 0, the method blocks indefinitely until a message arrives. For a simple client program, this may not matter. But if you do not want your program to consume system resources unnecessarily, use a timed synchronous receive.

Message message = messageConsumer.receive(1); // wait for 1 ms
Message message = messageConsumer.receiveNoWait(); // receive a message only if one is available

JMS Unified Client API

package jms;

import java.util.Properties;

import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.naming.*;

/**
 * <code>ServiceLocator</code> encapsulates JNDI lookups to make it
 * easier to access JNDI-based resources (EJBs, DataSources,
 * JMS Destinations, and so on).
 *
 */
public class ServiceLocator {
	private static Properties environment = null;
	
	static { // can be found in <JBoss>/server/default/conf/jndi.properties
		environment = new Properties();
		environment.setProperty("java.naming.provider.url", "localhost:1099");
		environment.setProperty("java.naming.factory.initial",
				"org.jboss.naming.NamingContextFactory");
		environment.setProperty("java.naming.factory.url.pkgs",
				"org.jboss.naming:org.jnp.interfaces");
	}

    /**
     * Making the default (no arg) constructor private
     * ensures that this class cannnot be instantiated.
     */
    private ServiceLocator() {   }


    /**
     * Gets a JMS <code>ConnectionFactory</code> using the given JNDI name.
     *
     * @param jmsConnectionFactoryJndiName The JMS <code>ConnectionFactory</code>'s JNDI name.
     * @return ConnectionFactory The JMS <code>ConnectionFactory</code>.
     * @throws ServiceLocatorException If there are JNDI lookup problems.
     * @see javax.jms.ConnectionFactory
     */
    public static ConnectionFactory getJmsConnectionFactory(String jmsConnectionFactoryJndiName) 
    	throws ServiceLocatorException {
        ConnectionFactory jmsConnectionFactory = null;

        try {
            Context ctx = new InitialContext(environment);
            
            jmsConnectionFactory = (ConnectionFactory) ctx.lookup(jmsConnectionFactoryJndiName);
        } catch (ClassCastException cce) {
            throw new ServiceLocatorException(cce);
        } catch (NamingException ne) {
            throw new ServiceLocatorException(ne);
        }

        return jmsConnectionFactory;
    }

    /**
     * Gets a JMS <code>Destination</code> using the given JNDI name.
     *
     * @param jmsDestinationJndiName The JMS <code>Destination</code>'s JNDI name.
     * @return Destination The JMS <code>Destination</code>.
     * @throws ServiceLocatorException If there are JNDI lookup problems.
     * @see javax.jms.Destination
     */
    public static Destination getJmsDestination(String jmsDestinationJndiName) 
    	throws ServiceLocatorException {
        Destination jmsDestination = null;

        try {
            Context ctx = new InitialContext(environment);

            jmsDestination = (Destination) ctx.lookup(jmsDestinationJndiName);
        } catch (ClassCastException cce) {
            throw new ServiceLocatorException(cce);
        } catch (NamingException ne) {
            throw new ServiceLocatorException(ne);
        }

        return jmsDestination;
    }
}

• Create the InitialContext to access the JNDI tree.
• Perform a JNDI lookup.
• Cast the object returned from JNDI to the correct JMS object type (ConnectionFactory or Destination) .
• Throw a ServiceLocatorException that chains a low-level JNDI-related exception and contains a corresponding error message.

Execute the example

package jms.consumer;

import javax.jms.*;

/**
 * <code>JmsListener</code> is the message listener of JMS Messages.
 * It implements the specified <code>onMessage</code> method.
 */
public class JmsListener implements MessageListener {
 
 public JmsListener( ) { }
 
 /**
 * Receives a JMS message.
 * @param message a JMS message
 */
 public void onMessage( Message message ) {
   // retrieve and process message
   try {
     if ( message instanceof TextMessage ) {
	TextMessage textMessage = (TextMessage) message;
	System.out.println(textMessage.getText());
     } else {
	System.out.println( "Expecting " +
         "TextMessage object, received " + message.getClass().getName() );
     }
  } // process JMS exception from message
     catch ( JMSException jmsException ) {
        jmsException.printStackTrace();
  }
 } // end method onMessage
}

// 5. Create a Message Consumer tied to the Destination.
messageConsumer = session.createConsumer(destination);

// 6. Initialize and set message listener
messageConsumer.setMessageListener(new JmsListener());    

// 7. Receive a Message.
connection.start();

// Give it some time to consume the messages
try {
    Thread.sleep(3000);
} catch (InterruptedException e) {
    e.printStackTrace();
}

// 8. Tear down the Message Consumer, Session, and Connection.
//messageConsumer.close();
//session.close();
connection.close();

private static final String MY_TOPIC = "topic/MyTopic";

// 5. Create a Message Consumer tied to the Destination.
messageConsumer = session.createConsumer(destination);

// 2. Get a Connection from the ConnectionFactory.
connection = connectionFactory.createConnection("myPublisher","passwd"); // username, password found in JMS_USERS

// 2. Get a Connection from the ConnectionFactory.
connection = connectionFactory.createConnection("mySubscriber","passwd"); // username, password found in JMS_USERS
...
 // 5. Create a durable subscriber tied to the Destination.
messageConsumer = session.createDurableSubscriber((Topic)destination, "mySubscriber");  // the SUBNAME from JMS_SUBSCRIPTIONS table

Message Driven Beans

package jms.consumer;

//Java extension packages
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.*;

/**
 * <code>MessageDrivenBean</code> is the message listener of JMS Messages.
 * It implements the specified <code>onMessage</code> method.
 */
@MessageDriven(activationConfig =
{
  @ActivationConfigProperty(propertyName="destinationType", propertyValue="javax.jms.Queue"),
  @ActivationConfigProperty(propertyName="destination", propertyValue="queue/MyQueue")
})
public class MessageDrivenBean implements MessageListener {
 
 /**
 * Receives a JMS message.
 * @param message a JMS message
 */
 public void onMessage( Message message ) {
   // retrieve and process message
   try {
     if ( message instanceof TextMessage ) {
	TextMessage textMessage = (TextMessage) message;
	System.out.println(textMessage.getText());
     } else {
	System.out.println( "Expecting " +
         "TextMessage object, received " + message.getClass().getName() );
     }
  } // process JMS exception from message
     catch ( JMSException jmsException ) {
        jmsException.printStackTrace();
  }
 } // end method onMessage
}

$> jar cf MDB.ejb3 jms/consumer/MessageDrivenBean.class

22:39:28,655 INFO  [STDOUT] Hello JMS

Transactions

/**
 * <code>MessageDrivenBean</code> is the message listener of JMS Messages.
 * It implements the specified <code>onMessage</code> method.
 */
@TransactionManagement
@MessageDriven(activationConfig =
{
  @ActivationConfigProperty(propertyName="destinationType", propertyValue="javax.jms.Queue"),
  @ActivationConfigProperty(propertyName="destination", propertyValue="queue/MyQueue")
})
public class MessageDrivenBean implements MessageListener {

 /**
 * Receives a JMS message.
 * @param message a JMS message
 */
 @TransactionAttribute(REQUIRED)
 public void onMessage( Message message ) {
   // retrieve and process message

/**
 * <code>MessageDrivenBean</code> is the message listener of JMS Messages.
 * It implements the specified <code>onMessage</code> method.
 */
@TransactionManagement(BEAN)
@MessageDriven(activationConfig =
{
  @ActivationConfigProperty(propertyName="destinationType", propertyValue="javax.jms.Queue"),
  @ActivationConfigProperty(propertyName="destination", propertyValue="queue/MyQueue")
})
public class MessageDrivenBean implements MessageListener {

 /**
 * Receives a JMS message.
 * @param message a JMS message
 */
 @Resource javax.transaction.UserTransaction transaction;
 public void onMessage( Message message ) {
   // retrieve and process message
   transaction.update(); // Start the transaction
   ...
   transaction.commit(); // Assume transaction went well, commit 

Summary

In this article we gave a detailed introduction to Java Messaging Service (JMS). JMS allows for anynchronous communication between applications in a loosely coupled manner. We discussed the JMS architecture, the messaging models (queue and topic), and we provided useful examples to help you up and running. We also used JBoss as our JMS provider, but you can use any other JMS Provider you wish. Please notice that we used the latest JMS 1.1 specification which allows for more generic code to be used. With JMS 1.1, the destination (queue or topic) is now simply a configuration setting, and you saw the impact in your code. 

There are some topics we didn’t cover in detail, like security, message selection, queue browsers etc. For that reason the interested reader is invited to check the references for more information.

References

1. SUN Microsystems (2002), JMS Specification JSR 914, http://java.sun.com/products/jms/docs.html.
2. Farrell, W. (2004), Introducing the Java Messaging Service, IBM.
3. Davis, S. & Marrs, T. (2005), JBoss at Work: A Practical Guide, O'Reilly.
4. Monson-Haefel, R. & Chappell D. A. (2001), Java Messaging Service, O'Reilly.
5. Haase, K. (2002), Java Messaging Service API Tutorial, SUN Microsystems.
6. Giotta, P. et. al. (2000), Professional JMS Programming, Apress.
7. JBoss (2006), Getting Started with JBoss 4.0, Release 5, JBoss Inc..
8. JBoss (2005), The JBoss 4 Application Server Guide, Release 3, JBoss Inc.
9. JBoss (2007), JBoss Messaging 1.4 User's Guide, JBoss Inc.
10. Jendrock E. et al. (2006), The Java™ EE 5 Tutorial, Third Edition, Addison Wesley Professional.
11. "JMS Transaction with JTA in JBoss 4", http://www.odi.ch/prog/jms-tx.php.
12. Mukhar K. & Zelenak C. (2006), Beginning Java EE 5 from Novice to Professional, Apress.

Creator: John N. Kostaras - email [email protected]

Last modification: 25 June 2008

URL: http://www.geocities.com/jnkjavaconnection/jmstutorial.html
URL: http://jkost.ergoway.gr/jnkjavaconnection/jmstutorial.html