Java Messaging Service (MOM- message oriented middleware)
Another aspect of flexibility is that Application B may not even be executing when Application A sends its message. The MOM will hold the message until Application B begins execution and attempts to retrieve its messages. This also prevents Application A from blocking while it waits for Application B to receive the message.
This asynchronous communication requires applications to be designed somewhat differently than most are designed today, but it can be an extremely useful method for
Time-independent or parallel processing.
Publish and subscribe:
Originally, enterprise messaging systems were developed to implement a point-to-point
model (PTP) in which each message produced by an application is received by one other application. In recent years, a new model has emerged, called publish and subscribe (or pub/sub).
Pub/sub replaces the single destination in the PTP model with a content hierarchy, known as topics. Sending applications publish their messages, indicating that the message represents information about a topic in the hierarchy.
Multiple applications may both subscribe and publish messages to a topic, and the
applications remain anonymous to each other. The MOM acts as a broker, routing the
published messages for a topic to all subscribers for that topic.
JMS:
JMS is a set of interfaces and associated semantics that define how a JMS client accesses the facilities of an enterprise messaging product.
Prior to JMS, each MOM vendor provided application access to their product through a
proprietary API, often available in multiple languages, including the Java language. JMS
provides a standard, portable way for Java programs to send and receive messages through a MOM product. Programs written with JMS will be able to run on any MOM that implements the JMS standard.
The key to JMS portability is the fact that the JMS API is provided by Sun as a set of
interfaces. Products that want to provide JMS functionality do so by supplying a provider that implements these interfaces.
JMS objectives:
The objectives of JMS, as stated in the specification, are to:
* Define a common set of messaging concepts and facilities.
* Minimize the concepts a programmer must learn to use enterprise messaging.
* Maximize the portability of messaging applications.
* Minimize the work needed to implement a provider.
* Provide client interfaces for both point-to-point and pub/sub domains. "Domains" is the
JMS term for the messaging models discussed earlier. (Note: A provider need not
implement both domains.)
What JMS does not provide
The following features, common in MOM products are not addressed by the JMS
specification. While acknowledged by the JMS authors as important for the development of robust messaging applications, these features are considered JMS provider-specific.
JMS providers are free to implement these features in any manner they please, if at all:
* Load balancing and fault tolerance
* Error and advisory system messages and notification
* Administration
* Security
* Wire protocol
* Message type repository
JMS overview and architecture:
Applications
A JMS application is comprised of the following elements:
* JMS clients. Java programs that send and receive messages using the JMS API.
* Non-JMS clients. It is important to realize that legacy programs will often be part of an
overall JMS application and their inclusion must be anticipated in planning.
* Messages. The format and content of messages to be exchanged by JMS and
non-JMS clients is integral to the design of a JMS application.
* JMS provider. As was stated previously, JMS defines a set of interfaces for which a
provider must supply concrete implementations specific to its MOM product.
* Administered objects. An administrator of a messaging system provider creates
objects that are isolated from the proprietary technologies of the provider.
Administered objects:
Providers of MOM products differ significantly in the mechanisms and techniques they use to implement messaging. To keep JMS clients portable, objects that implement the JMS interfaces have to be isolated from the proprietary technologies of a provider.
The mechanism for doing this is administered objects. These objects, which implement JMS interfaces, are created by an administrator of the provider's messaging system and are placed in the JNDI namespace.
The objects are then retrieved by JMS programs and accessed through the JMS interfaces
that they implement. The JMS provider must supply a tool that allows creation of
administered objects and their placement in the JNDI namespace.
There are two types of administered objects:
* ConnectionFactory: Used to create a connection to the provider's underlying
messaging system.
* Destination: Used by the JMS client to specify the destination of messages being
sent or the source of messages being received.
While the administered objects themselves are instances of classes specific to a provider's
implementation, they are retrieved using a portable mechanism (JNDI) and accessed through
portable interfaces (JMS). The JMS program only needs to know the JNDI name and the
JMS interface type of the administered object; no provider-specific knowledge is required.
Interfaces:
JMS defines a set of high-level interfaces that encapsulate various messaging concepts. In turn, these interfaces are further defined and customized for the two messaging domains --PTP and pub/sub.
* ConnectionFactory: An administered object that creates a Connection.
* Connection: An active connection to a provider.
* Destination: An administered object that encapsulates the identity of a message
destination, such as where messages are sent to or received from.
* Session: A single-threaded context for sending and receiving messages. For reasons
of simplicity and because Sessions control transactions, concurrent access by multiple
threads is restricted. Multiple Sessions can be used for multithreaded applications.
* MessageProducer: Used for sending messages.
* MessageConsumer: Used for receiving messages.
High-level interface PTP domain Pub/sub domain
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver, TopicSubscriber
QueueBrowser
Developing a JMS program:
A typical JMS program goes through the following steps to begin producing and consuming messages.
1. Look up a ConnectionFactory through JNDI.
2. Look up one or more Destinations through JNDI.
3. Use the ConnectionFactory to create a Connection.
4. Use the Connection to create one or more Sessions.
5. Use a Session and a Destination to create the required MessageProducers and MessageConsumers.
6. Start the Connection.
Messages:
JMS provides several
Message types for different types of content, but all messages derive from the Message
Interface.
A Message is divided into three constituent parts:
* The header is a standard set of fields that are used by both clients and providers to
identify and route messages.
* Properties provide a facility for adding optional header fields to a message. If your
application needs to categorize or classify a message in a way not provided by the
standard header fields, you can add a property to the message to accomplish that
categorization or classification. set
Property(...) methods are provided to set and get properties of a variety of Java
types, including Object. JMS defines a standard set of properties that are optional for
providers to supply.
* The body of the message contains the content to be delivered to a receiving
application. Each message interface is specialized for the type of content it supports.
Header fields:
The following list gives the name of each header field of Message, its corresponding Java
type, and a description of the field.
* JMSMessageID -- type string
Uniquely identifies each message that is sent by a provider. This field is set by the
provider during the send process; clients cannot determine the JMSMessageID for a
message until after it has been sent.
* JMSDestination -- type Destination
The Destination to which the message was sent; set by the provider during the send
process.
* JMSDeliveryMode -- type int
Contains the value DeliveryMode.PERSISTENT or
DeliveryMode.NON_PERSISTENT. A persistent message is delivered "once and only
once"; a non-persistent message is delivered "at most once." Be aware that "at most
once" includes not being delivered at all. A non-persistent message may be lost by a
provider during application or system failure. Extra care will be taken to assure that a
persistent message is not affected by failures. There is often considerable overhead in
sending persistent messages, and the trade-offs between reliability and performance
must be carefully considered when deciding the delivery mode of a message.
* JMSTimestamp -- type long
The time that the message was delivered to a provider to be sent; set by the provider
during the send process.
* JMSExpiration -- type long
The time when a message should expire. This value is calculated during the send
process as the sum of the time-to-live value of the sending method and the current
time. Expired messages should not be delivered by the provider. A value of 0 indicates
that the message will not expire.
* JMSPriority -- type int
The priority of the message; set by the provider during the send process. A priority of 0
is the lowest priority; a priority of 9 is the highest priority.
* JMSCorrelationID -- type string
Typically used to link a response message with a request message; set by the JMS
program sending the message. A JMS program responding to a message from another
JMS program would copy the JMSMessageID of the message it is responding to into
this field, so that the requesting program could correlate the response to the particular
request that it made.
* JMSReplyTo -- type Destination
Used by a requesting program to indicate where a reply message should be sent; set by
the JMS program sending the message.
* JMSType -- type string
Can be used by a JMS program to indicate the type of the message. Some providers
maintain a repository of message types and will use this field to reference the type
definition in the repository; in this case, the JMS program should not use this field.
* JMSRedelivered -- type boolean
Indicates that the message was delivered earlier to the JMS program, but that the
program did not acknowledge its receipt; set by the provider during receive processing.
Standard properties:
The following list gives the name of each standard property of Message, its corresponding Java type, and a description of the property. Support for standard properties by a provider is optional. JMS reserves the "JMSX" property name for these and future JMS-defined properties.
* JMSXUserID -- type string
Identity of the user sending the message.
* JMSXApplID -- type string
Identity of the application sending the message.
* JMSXDeliveryCount -- type int
Number of times delivery of the message has been attempted.
* JMSXGroupID -- type string
Identity of the message group to which this message belongs.
* JMSXGroupSeq -- type int
Sequence number of this message within the message group.
* JMSXProducerTXID -- type string
Identity of the transaction within which this message was produced.
* JMSXConsumerTXID -- type string
Identity of the transaction within which this message was consumed.
* JMSXRcvTimestamp -- type long
The time JMS delivered the message to the consumer.
* JMSXState -- type int
Used by providers that maintain a message warehouse of messages; generally not of
interest to JMS producers or consumers.
* JMSX_
Reserved for provider-specific properties.
Message body:
There are five forms of message body, and each form is defined by an interface that extends
Message. These interfaces are:
* StreamMessage: Contains a stream of Java primitive values that are filled and read
sequentially using standard stream operations.
* MapMessage: Contains a set of name-value pairs; the names are of type string and
the values are Java primitives.
* TextMessage: Contains a String.
* ObjectMessage: Contains a Serializable Java object; JDK 1.2 collection classes
can be used.
* BytesMessage: Contains a stream of uninterpreted bytes; allows encoding a body to
match an existing message format.
Transactions:
A JMS transaction groups a set of produced messages and a set of consumed messages into an atomic unit of work. If an error occurs during a transaction, the production and consumption of messages that occurred before the error can be "undone."
Session objects control transactions and a Session may be denoted as transacted when it is created. A transacted Session always has a current transaction, that is, there is no begin(); commit() and rollback() end one transaction and automatically begin another.
Acknowledgement:
Acknowledgement is the mechanism whereby a provider is informed that a message has
been successfully received.
If the Session receiving the message is transacted, acknowledgement is handled
automatically. If the Session is not transacted, then the type of acknowledgement is
determined when the Session is created.
There are three types of acknowledgement:
* Session.DUPS_OK_ACKNOWLEDGE: Lazy acknowledgement of message delivery;
reduces overhead by minimizing work done to prevent duplicates; should only be used if
duplicate messages are expected and can be handled.
* Session.AUTO_ACKNOWLEDGE: Message delivery is automatically acknowledged upon
completion of the method that receives the message.
* Session.CLIENT_ACKNOWLEDGE: Message delivery is explicitly acknowledged by
calling the acknowledge() method on the Message.
JMS and XML:
The authors of JMS included the TextMessage message type on the presumption that
String messages will be used extensively.
Their reasoning is that XML will be a popular, if not the most popular, means of representing
the content of messages. A portable transport mechanism (JMS) coupled with a portable
data representation (XML) is proving to be a powerful tool in enterprise application
integration (EAI) and other areas of data exchange..
JMS and J2EE:
J2EE version 1.2 requires compliant application servers to have the JMS API present, but does not mandate the presence of a JMS provider.
J2EE version 1.3 will require application servers to supply a JMS provider.
Another important development in JMS capabilities is the message-driven bean of the EJB 2.0 specification , which will add asynchronous notification abilities to Enterprise JavaBeans containers. A message-driven bean, which will implement the MessageListener interface, will be invoked by the EJB container on the arrival of a
message at a destination designated at deployment time. The message-driven bean will
contain the business logic to process the message, including, if needed, the invoking of other enterprise beans
Point-to-point interfaces:
JMS interfaces for point-to-point programming and some of their methods.
QueueConnectionFactory:
QueueConnectionFactory is an administered object that is retrieved from JNDI to create a connection to a provider. It contains a createQueueConnection() method which returns a QueueConnection object.
QueueConnection
QueueConnection encapsulates an active connection to a provider. Some of its methods are:
* createQueueSession(boolean, int): Returns a QueueSession object. The
boolean parameter indicates whether the QueueSession is transacted or not; the int indicates the acknowledgement mode.
* start() (inherited from Connection): Activates the delivery of messages from the
provider.
* stop() (inherited from Connection): Temporarily stops delivery of messages;
delivery can be restarted with start().
* close() (inherited from Connection): Closes the connection to the provider and
releases all resources held in its behalf.
QueueSession
QueueSession is the single-threaded context for sending and receiving PTP messages.
Some of its methods are:
* createSender(Queue): Returns a QueueSender object to send messages to the
specified Queue.
* createReceiver(Queue): Returns a QueueReceiver object to receive messages
from the specified Queue.
* createBrowser(Queue): Returns a QueueBrowser object to browse messages on
the specified Queue.
* commit() (inherited from Session): Commits all consumed or produced messages for
the current transaction.
* rollback() (inherited from Session): Rolls back all consumed or produced
messages for the current transaction.
* create
methods that return a
Queue
Queue encapsulates a point-to-point destination. It is an administered object that is
retrieved from JNDI.
QueueSender
QueueSender is used to send point-to-point messages. Some of its methods are:
* send(Message): Sends the indicated Message.
* setDeliveryMode(int) (inherited from MessageProducer): Sets the delivery
mode for subsequent messages sent; valid values are DeliveryMode.PERSISTENT
and DeliveryMode.NON_PERSISTENT.
* setPriority(int) (inherited from MessageProducer): Sets the priority for
subsequent messages sent; valid values are 0 through 9.
* setTimeToLive(long) (inherited from MessageProducer): Sets the duration before expiration, in milliseconds, of subsequent messages sent.
QueueReceiver
QueueReceiver is used to receive point-to-point messages. Some of its methods are:
* receive() (inherited from MessageConsumer): Returns the next message that
arrives; this method blocks until a message is available.
* receive(long) (inherited from MessageConsumer): Receives the next message that
arrives within long milliseconds; this method returns null if no message arrives within
the time limit.
* receiveNoWait (inherited from MessageConsumer): Receives the next message if
one is immediately available; this method returns null if no message is available.
* setMessageListener(MessageListener) (inherited from MessageConsumer):
Sets the MessageListener; the MessageListener object receives messages as
they arrive, that is, asynchronously.
QueueBrowser
When using QueueReceiver to receive messages, the messages are removed from the queue when they are received. QueueBrowser is used to look at messages on a queue without removing them. The method for doing that is getEnumeration(), which returns a java.util.Enumeration that can be used to scan the messages in the queue; changes to the queue (arriving and expiring of messages) may or may not be visible.
MessageListener
MessageListener is an interface with a single method -- onMessage(Message) – that provides asynchronous receipt and processing of messages.
This interface should be implemented by a client class and an instance of that class passed to the QueueReceiver object with the setMessageListener(MessageListener)
method. As a message arrives on a queue, it is passed to the object by calling the
onMessage(Message) method.MessageListener objects are used in both the PTP and pub/sub domains.
Point-to-point programming:
Create these two Java Files (QSender.java and QReceiver.java.)
import java.io.*;
import javax.jms.*;
import javax.naming.*;
public class QSender {
public static void main(String[] args) {
new QSender().send();
}
public void send() {
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
//Prompt for JNDI names
System.out.println("Enter QueueConnectionFactory name:");
String factoryName = reader.readLine();
System.out.println("Enter Queue name:");
String queueName = reader.readLine();
//Look up administered objects
InitialContext initContext = new InitialContext();
QueueConnectionFactory factory =
(QueueConnectionFactory) initContext.lookup(factoryName);
Queue queue = (Queue) initContext.lookup(queueName);
initContext.close();
//Create JMS objects
QueueConnection connection = factory.createQueueConnection();
QueueSession session =
connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueSender sender = session.createSender(queue);
//Send messages
String messageText = null;
while (true) {
System.out.println("Enter message to send or 'quit':");
messageText = reader.readLine();
if ("quit".equals(messageText))
break;
TextMessage message = session.createTextMessage(messageText);
sender.send(message);
}
//Exit
System.out.println("Exiting...");
reader.close();
connection.close();
System.out.println("Goodbye!");
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
}
}
QReceiver File:
import java.io.*;
import javax.jms.*;
import javax.naming.*;
public class QReceiver implements MessageListener {
private boolean stop = false;
public static void main(String[] args) {
new QReceiver().receive();
}
public void receive() {
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
//Prompt for JNDI names
System.out.println("Enter QueueConnectionFactory name:");
String factoryName = reader.readLine();
System.out.println("Enter Queue name:");
String queueName = reader.readLine();
reader.close();
InitialContext initContext = new InitialContext();
QueueConnectionFactory factory =
(QueueConnectionFactory) initContext.lookup(factoryName);
Queue queue = (Queue) initContext.lookup(queueName);
initContext.close();
//Create JMS objects
QueueConnection connection = factory.createQueueConnection();
QueueSession session =
connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
QueueReceiver receiver = session.createReceiver(queue);
receiver.setMessageListener(this);
connection.start();
//Wait for stop
while (!stop) {
Thread.sleep(1000);
}
//Exit
System.out.println("Exiting...");
connection.close();
System.out.println("Goodbye!");
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
}
public void onMessage(Message message) {
try {
String msgText = ((TextMessage) message).getText();
System.out.println(msgText);
if ("stop".equals(msgText))
stop = true;
} catch (JMSException e) {
e.printStackTrace();
stop = true;
}
}
}
Running the programs
you will need the javax.naming and javax.jms packages to compile these programs.Before you run these programs, you'll need to use the administration tool supplied by your JMS provider to create the QueueConnectionFactory and Queue administered objects and place them in the JNDI namespace.
You also need to make sure that the provider's JMS implementation classes are on your
classpath. You can then run both of these programs at the same time, supplying the same JNDI names for the QueueConnectionFactory and Queue, and send messages from the QSender to the QReceiver.
Pub/sub interfaces:
TopicConnectionFactory:
TopicConnectionFactory is an administered object that is retrieved from JNDI in order to create a connection to a provider. It contains a createTopicConnection() method which returns a TopicConnection object.
TopicConnection:
TopicConnection encapsulates an active connection to a provider. Some of its methods are:
* createTopicSession(boolean, int): Returns a TopicSession object. The
boolean parameter indicates whether the TopicSession is transacted; the int
indicates the acknowledgement * start() (inherited from Connection): Activates the delivery of messages from the
provider.
* stop() (inherited from Connection): Temporarily stops delivery of messages;
delivery can be restarted with start().
* close() (inherited from Connection): Closes the connection to the provider and
releases all resources held in its behalf
TopicSession
TopicSession is the single-threaded context for sending and receiving pub/sub messages.
Some of its methods are:
* createPublisher(Topic): Returns a TopicPublisher object to send messages
to the specified Topic.
* createSubscriber(Topic): Returns a TopicSubscriber object to receive
messages from the specified Topic. This subscriber is non-durable ; that is, the
subscription will only last for the lifetime of the object and will only receive messages
when it is active.
* createDurableSubscriber(Topic, String): Returns a TopicSubscriber
object to receive messages from the specified Topic, giving the String name to the
subscriber. Messages for a durable subscriber will be retained by JMS if the object is
not active and will be delivered to subsequent subscriber objects that are created with
the same name.
* unsubscribe(String): Ends the subscription with the String name.
* commit() (inherited from Session): Commits all consumed or produced messages for the current transaction.
* rollback() (inherited from Session): Rolls back all consumed or produced
messages for the current transaction.
* create
methods that return a
TextMessage, and so on.
Topic
Topic encapsulates a pub/sub destination. It is an administered object that is retrieved from JNDI.
TopicPublisher
TopicPublisher is used to send pub/sub messages. Some of its methods are:
* publish(Message): Publishes the indicated Message.
* setDeliveryMode(int) (inherited from MessageProducer): Sets the delivery
mode for subsequent messages sent; valid values are DeliveryMode.PERSISTENT
and DeliveryMode.NON_PERSISTENT.
* setPriority(int) (inherited from MessageProducer): Sets the priority for
subsequent messages sent; valid values are 0 through 9.
* setTimeToLive(long) (inherited from MessageProducer): Sets the duration before
expiration, in milliseconds, of subsequent messages sent
TopicSubscriber
TopicSubscriber is used to receive point-to-point messages. Some of its methods are:
* receive() (inherited from MessageConsumer): Returns the next message that
arrives; this method blocks until a message is available.
* receive(long) (inherited from MessageConsumer): Receives the next message that
arrives within long milliseconds; this method returns null if no message arrives within
the time limit.
* receiveNoWait (inherited from MessageConsumer): Receives the next message if
one is immediately available; this method returns null if no message is available.
* setMessageListener(MessageListener) (inherited from MessageConsumer):
Sets the MessageListener; the MessageListener object receives messages as
they arrive, that is, asynchronously.
Pub/sub programming:
Create Java File:
TPublisher.java
import java.io.*;
import javax.jms.*;
import javax.naming.*;
public class TPublisher {
public static void main(String[] args) {
new TPublisher().publish();
}
public void publish() {
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
//Prompt for JNDI names
System.out.println("Enter TopicConnectionFactory name:");
String factoryName = reader.readLine();
System.out.println("Enter Topic name:");
String topicName = reader.readLine();
//Look up administered objects
InitialContext initContext = new InitialContext();
TopicConnectionFactory factory =
(TopicConnectionFactory) initContext.lookup(factoryName);
Topic topic = (Topic) initContext.lookup(topicName);
initContext.close();
//Create JMS objects
TopicConnection connection = factory.createTopicConnection();
TopicSession session =
connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicPublisher publisher = session.createPublisher(topic);
//Send messages
String messageText = null;
while (true) {
System.out.println("Enter message to send or 'quit':");
messageText = reader.readLine();
if ("quit".equals(messageText))
break;
TextMessage message = session.createTextMessage(messageText);
publisher.publish(message);
}
//Exit
System.out.println("Exiting...");
reader.close();
connection.close();
System.out.println("Goodbye!");
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
}
}
CREATE TSubscriber.java File
import java.io.*;
import javax.jms.*;
import javax.naming.*;
public class TSubscriber implements MessageListener {
private boolean stop = false;
public static void main(String[] args) {
new TSubscriber().subscribe();
}
public void subscribe() {
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
try {
//Prompt for JNDI names
System.out.println("Enter TopicConnectionFactory name:");
String factoryName = reader.readLine();
System.out.println("Enter Topic name:");
String topicName = reader.readLine();
reader.close();
//Look up administered objects
InitialContext initContext = new InitialContext();
TopicConnectionFactory factory =
(TopicConnectionFactory) initContext.lookup(factoryName);
Topic topic = (Topic) initContext.lookup(topicName);
initContext.close();
//Create JMS objects
TopicConnection connection = factory.createTopicConnection();
TopicSession session =
connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
TopicSubscriber subscriber = session.createSubscriber(topic);
subscriber.setMessageListener(this);
connection.start();
//Wait for stop
while (!stop) {
Thread.sleep(1000);
}
//Exit
System.out.println("Exiting...");
connection.close();
System.out.println("Goodbye!");
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
}
public void onMessage(Message message) {
try {
String msgText = ((TextMessage) message).getText();
System.out.println(msgText);
if ("stop".equals(msgText))
stop = true;
} catch (JMSException e) {
e.printStackTrace();
stop = true;}
}
}
