|
|
[an error occurred while processing this directive]
|
|
|
Messaging systems are used to build reliable, scalable, high-performance applications. A messaging system shields programmers from the complications of message buffering and dissemination, crash recovery, client registrations, lost connections, and more. However, messaging systems have so far been proprietary, making application code non-portable and difficult to migrate from product to product. Also, without common concepts and APIs, the learning curve for messaging systems has been steep.
The Java Message Service (JMS) is an API defined by Sun Microsystems that specifies how Java client applications can access an enterprise messaging system. JMS plays a fundamental role in J2EE as the foundation for Message Driven Beans. Although different vendors will still implement JMS using different transport mechanisms and message protocols, the specification allows programmers to write portable code using familiar concepts. This article introduces JMS and provides sample code for using the API.
 |
the Java Message Service |
|
|
|
Figure 1 shows the main concepts in JMS. The right side illustrates the point-to-point (P2P) model, and the left side illustrates the publish/subscribe (pub/sub) model. In P2P messaging, the message senders and receivers are de-coupled by a queue. In pub/sub messaging, the publishers and subscribers are de-coupled by a topic. Although the pub/sub and P2P communication models are different from a conceptual point of view, they maintain a high degree of commonality; in JMS, which centers on a generic-messaging model, pub/sub and P2P are merely specializations.
Figure 1: The main concepts in JMS.
In general, JMS applications have to perform the following five steps in order to start sending or receiving messages:
- Resolve a connection factory and a destination from JNDI. A pub/sub application will resolve a TopicConnectionFactory, whereas a P2P application will resolve a QueueConnectionFactory. A Destination is either a Queue or a Topic.
- Create a connection using the connection factory. Again, a pub/sub application needs to create a TopicConnection, and a P2P application needs to create a QueueConnection.
- Use the connection to create a session with the desired properties (TopicSession for pub/sub and QueueSession for P2P).
- If the application is a supplier, create a message producer (MessageProducer is the base interface for QueueSender and TopicPublisher). If the application is a consumer, create a message consumer (MessageConsumer is the base interface for QueueReceiver and TopicSubscriber).
- Start to send or receive messages using the producer or consumer object. A producer will use the session to create different kinds of messages.
The database in Figure 1 illustrates that JMS will store persistent messages for recovery purposes in some store. Some JMS vendors might use a file-based store instead of a database. Typically, JMS doesn't use a database when handling non-persistent messages.
|
|
|
message model |
|
|
|
JMS defines five standard messages that are used to carry different kinds of payloads. All message types derive from the Message interface to provide a common header. The message header contains various quality-of-service (QoS) properties as well as user-defined properties. Some of the more important QoS properties include reliability (transient or persistent), priority, and expiration. The other five message types are:
- MapMessage supports the Message interface and provides a body of name/value pairs.
- BytesMessage supports a body with un-interpreted data. The message supports the methods of the DataInputStream and DataOutputStream interfaces from the Java I/O package.
- ObjectMessage provides a body that can contain any Java object that supports the Serializable interface.
- StreamMessage adds a body that is a stream of objects.
- TextMessage provides a body that is a Java String.
An important concept in JMS is message acknowledgement. The default acknowledgement mode is AUTO_ACKNOWLEDGE, which means that the messaging system automatically acknowledges messages on behalf of the consumer. With automatic message acknowledge, a consumer will never receive duplicate messages, subject to the limitations of distributed systems. JMS also supports CLIENT_ACKNOWLEDGE, which implies that the consumer is responsible for acknowledging messages. Finally, there is DUPS_OK_ACKNOWLEDGE, which provides better performance at the cost of potentially receiving duplicate messages in the event of a failure.
|
|
|
publish/subscribe |
|
|
|
|
As described above, the pub/sub model centers on a topic. Publishers send messages to a topic, and all subscribers that are connected to the topic in question will receive the messages (unless their selector doesn't match the message). A topic subscriber will not receive any messages that were sent to the topic prior to the time of connection. A durable subscriber is a special kind of subscriber that can be inactive at times. Once a durable subscriber is re-started, it will receive all messages that were sent to its topic since the time that it became inactive.
|
|
|
point-to-point |
|
|
|
|
The P2P model centers on a queue. A queue is different from a topic in that it will retain messages even if there are no receivers at the time the messages are sent to the queue. Also, once a message has been consumed from a queue, it will not be delivered to any other consumers. Although the JMS specification doesn't define how a queue should behave if more than one receiver is connected, a typical behavior would be to round robin messages from the queue to the receivers. Portable applications should only have one receiver per queue.
|
|
|
sending messages |
|
|
|
|
JMS applications create messages using a session and send them to a Destination using a MessageProducer. Having described the concepts above, this section shows Java example code for sending messages in the P2P and pub/sub models. Note that most of the methods in the JMS API can raise a JMSException, but for simplicity, the examples below do not show exception handling.
A Queue Sender
In the P2P model, senders use QueueSender, which is derived from the MessageProducer interface. Example 1 shows the five steps described above for a queue sender.
InitialContext ctx = new InitialContext();
Queue queue = (Queue) ctx.lookup("queue/queue0");
QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup("queue/connectionFactory");
QueueConnection queueConn = connFactory.createQueueConnection();
QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
QueueSender queueSender = queueSession.createSender(queue);
TextMessage message = queueSession.createTextMessage();
message.setText("hello world"); queueSender.send(message);
Example 1: Code for a simple queue sender.
All classes and interfaces are from the javax.jms package, except InitialContext. The following example code should be self-explanatory:
- Resolve the QueueConnectionFactory and a Queue from JNDI.
- Create a QueueConnection.
- Create a QueueSession. The first argument specifies whether the session is transacted (as described below). The second argument for message acknowledgement is not used by message producers.
- Create a QueueSender for the queue.
- Create and send a text message with no properties.
The connection factory and the destinations are administrated objects, which will be created using a JMS provider's tools (as explained later in this article). A connection is typically a heavyweight object that represents some physical connection to a JMS server. As a guideline, applications should therefore try to avoid creating too many connections. A session represents a single-threaded object, which handles incoming and outgoing messages for a set of producers and consumers.
The QueueSender interface supports four different send methods with different parameters. In a non-transacted session, calling send implies that the message has been received by JMS. In a transacted session, JMS merely buffers messages when send is called, and the messages are not sent until the commit method is called on the session. A transacted session also supports a rollback method, which removes all messages sent by the sender.
The MessageProducer base interface supports a number of methods to set default properties for the messages sent by a producer. By default all messages sent by JMS are persistent and will survive failures of the JMS server at some additional cost. To make all messages transient rather than persistent, the following can be used:
QueueSender queueSender = queueSession.createSender(queue);
queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
Example 2: Making transient messages the default for a QueueSender.
Most JMS objects like connection, session, producer, and consumer have a close operation that instructs JMS when the client application no longer requires the object in question. So rather than relying on garbage collection, it is recommended to use the close operation when the application is done with an object. This allows JMS to free the resources allocated on its behalf.
A Topic Publiser
Sending messages to a topic rather than a queue requires almost no code changes. The code shown in Example 3 creates a TopicPublisher by replacing all occurrences of 'queue' (as found in Example 1) with 'topic' and renaming the sender to a publisher.
InitialContext ctx = new InitialContext();
Topic topic = (Topic) ctx.lookup("topic/topic0");
TopicConnectionFactory connFactory = (TopicConnectionFactory) ctx.lookup("topic/connectionFactory");
TopicConnection topicConn = connFactory.createTopicConnection();
TopicSession topicSession = topicConn.createTopicSession(false, Session.DUPS_OK_ACKNOWLEDGE);
TopicPublisher topicPublisher = topicSession.createPublisher(topic);
Example 3: Code for a simple topic publisher.
Recall that all messages have a header section for user-defined properties, which consumers can use for filtering. To publish stock quotes on a topic, a topic publisher could do the following:
Message quote = topicSession.createMessage();
quote.setFloatProperty("SSSW", 18.23);
topicPublisher.publish(quote);
Example 4: Publishing stock quotes on a topic.
|
|
|
receiving messages |
|
|
|
|
JMS applications can receive messages from a destination using a synchronous polling model or an asynchronous callback model. The base interface for consumers is MessageConsumer. The following sections describe the different kinds of consumers supported by JMS.
A Queue Receiver
A QueueReceiver is the P2P interface used to consume messages from a queue. Example 5 shows a queue receiver that uses a non-transacted session with automatic message acknowledge to synchronously receive a single message from a queue.
InitialContext ctx = new InitialContext();
Queue queue = (Queue) ctx.lookup("queue/queue0");
QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup("queue/connectionFactory");
QueueConnection queueConn = connFactory.createQueueConnection();
QueueSession queueSession = queueConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueReceiver queueReceiver = queueSession.createReceiver(queue);
queueConn.start(); Message message = queueReceiver.receive();
Example 5: Synchronous receive of a single message from a queue.
Note that a message consumer doesn't start to receive messages until the start method of the connection has been called. This allows consumers to initialize properly before any messages will be delivered by the session. There are three different receive methods for synchronous message consumption:
- The receive method with no arguments waits indefinitely for a message to arrive.
- The receive method with a timeout gets the next message that arrives within the specified timeout interval.
- The receiveNoWait method returns immediately with a message or null if no messages are currently available on the destination.
Any attempt to receive a message before a connection has been started or after a connection has been stopped will be blocked.
Rather than having messages automatically acknowledged by JMS, a consumer can also use manual acknowledgements as follows:
QueueConnection queueConn = connFactory.createQueueConnection();
QueueSession queueSession = queueConn.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
QueueReceiver queueReceiver = queueSession.createReceiver(queue);
Example 6: Creating a queue receiver with manual (client) acknowledgement.
Consumer applications that use client acknowledgement must use the acknowledge method on the Message interface to inform the JMS server that it has successfully consumed the message (see Example 7). As a guideline, consumer applications should acknowledge messages as quickly as possible to prevent JMS from using too many resources to keep track of unacknowledged messages. Note that if a consumer uses a transacted session, it should instead use the commit method of the session to acknowledge messages.
Message mesage = queueReceiver.receiveNoWait();
if (message != null) {
// do something with the message ...
message.acknowledge(); }
Example 7: Using the acknowledge method to manually acknowledge a message.
Queue receivers can also consume messages asynchronously using a message listener. The MesssgeListener interface is briefly described in the next section. Since JMS maintains a high degree of commonality between pub/sub and P2P, synchronous and asynchronous message receipt is identical in both models.
A Topic Subscriber
In pub/sub consumer applications the TopicSubscriber interface is used to receive messages. As described above, a topic subscriber can also use the receive method to receive consumer messages synchronously. In order to use the asynchronous model, a consumer must register a callback object that implements the MessageListener interface as seen in the following example:
InitialContext ctx = new InitialContext();
Topic topic = (Topic) ctx.lookup("topic/topic0");
TopicConnectionFactory connFactory = (TopicConnectionFactory) ctx.lookup("topic/connectionFactory");
TopicConnection topicConn = connFactory.createTopicConnection();
TopicSession topicSession = topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber topicSubscriber = topicSession.createSubscriber(topic);
topicSubscriber.setMessageListener(new MyListener()); topicConn.start();
Example 8: TopicSubscriber that registers an asynchronous message listener.
The MessageListener interface has a single method onMessage that must be implemented. The consumer class itself or an inner class can conveniently be used for that purpose as follows:
public class MyListener implements MessageListener {
public void onMessage(Message message) {
System.out.println("SSSW quote: " + message.getFloatProperty("SSSW"));
}
}
Example 9: Class that implements the MessageListener interface.
When using synchronous receives, an application will be notified of any problems when the receive method throws a JMSException. This is not possible when using the asynchronous MessageListener interface. JMS supports an ExceptionListener, which may be registered with the connection. It is recommended that asynchronous consumers use this facility.
|
|
|
message selectors |
|
|
|
Message selectors are used for two things:
- To allow consumers to receive only the messages in which they are interested. (This avoids excessive message processing in the consumer, and in turn provides more responsive consumer applications.)
- To limit the amount of messages that the JMS must send to its consumers. (In many messaging applications the network bandwidth can become bottlenecked; selectors can be used to relieve this problem.)
JMS uses an SQL like grammar for selecting messages. It is only possible to write selectors for the user-defined properties in the header of a message and for some of the system-defined header properties. The selector is set when the consumer is created from the session as indicted below:
TopicSession topicSession = topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber topicSubscriber = topicSession.createSubscriber(topic, "SSSW 25");
Example 10: Setting a selector when creating a topic subscriber.
In the above example, the subscriber will only receive messages where the 'SSSW' property is greater than 25. Table 1 lists more sample selectors.
| Selector expression | Description |
| JMSType = 'quote' SSSW 25 | Identifiers can either be header-field references or user-defined property references. |
| SSSW 25 AND CSCO 50 | The AND, OR and NOT operators can be used to form logical expressions. As in Java, the =, , =, <, <= and < comparison operators can be used. |
| (length * 17 + height / 42) * 5 = 3 | Java +, -, * and / can be used for arithmetic operations. Parenthesis can be used to order expressions. |
| SSSW BETWEEN 15 AND 25 | Same as: SSSW = 15 AND SSSW <= 25. |
| QUOTE IN ('SSSW' 'CSCO') | Same as: (QUOTE = 'SSSW') OR (QUOTE = 'CSCO'). |
| SSSW IS NULL | True if there is no property named 'SSSW' in the header. |
| number LIKE '12%3' | True for '123' and '12443' but false for '1234'. |
| word LIKE 'h_llo' | True for 'hello' but false for 'heeello'. |
Table 1: Sample selector expressions using the SQL grammar.
Once a message selector has been set on a consumer it is not possible to change it. However, a consumer can always disconnect and then reconnect to the same destination with a new selector.
|
|
|
durable subscribers |
|
|
|
Durable subscribers are used for topics with persistent messages. A durable subscriber can be inactive at times. JMS holds messages received by the topic while the subscriber is inactive and delivers them when the subscriber re-subscribes. This means that if the machine with the consumer application crashes, the durable subscriber will not loose any messages. Durable subscribers are created using the following session:
TopicSession topicSession = topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber topicSubscriber = topicSession.createDurableSubscriber(topic, "name");
Example 11: Creating a durable topic subscriber.
A durable subscriber is identified by a name. The first time a durable subscriber is created, JMS will register a new subscriber for the topic and start sending messages to it. A durable subscriber becomes active in the same manner. In other words, if the createDurableSubscriber method is called with an existing subscriber name, JMS will not create a new subscriber, but will simply re-start the message flow from the topic to the now active durable subscriber.
Durable subscribers take up more resources than regular, transient subscribers because JMS has to store messages on behalf of the subscriber even if it is inactive. Therefore, it is important to unsubscribe durable subscribers when they are no longer required by the client application. This prevents the JMS server from buffering up unnecessary messages, and can be accomplished by using the unsubscribe method of the session as follows:
topicSession.unsubscribe("name");
Example 12: Unsubscribing a durable subscriber.
Once unsubscribed, the messages held on behalf of the subscriber will be permanently destroyed. Durable subscribers are otherwise identical to regular subscribers, i.e. they can specify selectors and use sessions with different acknowledge modes.
|
|
|
queue browsers |
|
|
|
A queue browser is a special kind of queue receive that can look at messages on a queue without actually consuming and removing them. For instance, if a queue contains messages that represent some kind of work, consumer applications might want to use a queue browser to decide whether they wish to perform any of those tasks. A queue browser uses an iterator pattern to inspect the contents of a queue as follows:
InitialContext ctx = new InitialContext();
Queue queue = (Queue) ctx.lookup("queue/queue0");
QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup("queue/connectionFactory");
QueueConnection queueConn = connFactory.createQueueConnection();
QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
QueueBrowser queueBrowser = queueSession.createBrowser(queue);
Enumeration enum = queueBrowser.getEnumeration();
while (enum.hasMoreElements()) {
Message message = (Message) enum.nextElement();
System.out.println("message type: " + message.getJMSType());
}
queueBrowser.close();
Example 13: Using the QueueBrowser interface to inspect a queue.
A QueueBrowser can also have a message selector to limit or apply a view on the messages that are returned in the enumeration.
|
|
|
administration |
|
|
|
|
It is important to realize that JMS is a client API. It only defines how a Java client accesses the messaging capabilities of an enterprise messaging system. Other important aspects of a messaging system, including administration and security, are not defined by JMS. Each JMS vendor will have proprietary tools for administration of the messaging system. This section provides a brief description of jBroker MQ, a 100% Java implementation of JMS on top of the jBroker ORB.
|
|
|
jBroker MQ |
|
|
|
Compared to many other JMS implementations, jBroker MQ is a designated JMS server. It was designed and built with JMS in mind, and it doesn't require inefficient layers of wrapper code to convert the JMS client API into a proprietary API of an existing messaging middleware product. As mentioned earlier, JMS is a client side API and doesn't standardize important server-side aspects of the message service. Below is a summary of the implementation features of jBroker MQ:
- Wire protocol: jBroker MQ is built on top of jBroker, which is an enterprise ORB supporting the CORBA 2.3 specification. The wire protocol for sending messages from the JMS server to its clients is therefore IIOP. IIOP is a robust and proven protocol for communicating between distributed objects executing in a heterogeneous environment.
- Administration: jBroker MQ supports APIs for creating and destroying queues and topics. JMS administrated objects, along with any other persistent information destinations, are accessed at runtime through a standard COS Naming Service and their properties are stored in a database using JDBC.
- Security: jBroker MQ supports APIs for managing users and groups, and for managing Access Control Lists (ACLs). Standard roles for consuming messages, producing messages, and message administration are defined.
In summary, the server side of jBroker MQ is also standards-based wherever possible.
|
|
|
destination administration |
|
|
|
Although large-scale and production systems typically require administrators to define security policies, people who are familiar with JMS will typically perform most destination management processes. In jBroker MQ, creating and destroying destinations is performed using the connection implementation object. The following example creates 'queue0' and deletes 'queue1':
import javax.jms.*;
import javax.naming.*;
import com.sssw.jms.api.*;
import com.sssw.jms.api.admin.*;
public class CreateDestroy {
public static void main(String[] args) throws Exception {
InitialContext ctx = new InitialContext();
TopicConnectionFactory tfac = (TopicConnectionFactory) ctx.lookup("topic/connectionFactory");
JMQTopicConnection topicConn = (JMQTopicConnection) tfac.createTopicConnection();
JMQDestinationAdmin destAdmin = topicConn.getDestinationAdmin();
Topic topic = (Topic) destAdmin.createDestination("queue0", JMQDestination.TOPIC, null);
destAdmin.deleteDestination("queue1", JMQDestination.TOPIC);
topicConn.close();
}
}
Example 14: Creating and destroying a topic.
This example shows that once a connection object has been created it is possible to get a reference to an object that supports the JMQDestinationAdmin interface. This interface has two simple methods for destination management, namely createDestination and deleteDestination. The third argument in the createDestination method can optionally contain a Properties object, which defines properties for the queue or topic.
|
|
|
conclusion |
|
|
|
|
JMS is an important API because it provides simplified access to enterprise messaging systems from Java programs. Also, JMS is the API for messaging in J2EE, which has become the platform of choice for server side Java development. One of the major strengths of JMS is that it provides a common, simple-to-use messaging model while also offering many of the features that are expected from messaging middleware systems.
|
|
|
|
|
|
