Overview
The Java Message Service (JMS) allows Java applications to send and receive messages asynchronously. It can be used for communication between server applications, but also within an application to postpone tasks or to distribute them to components.
In order to use JMS, you usually need some sort of messaging provider. This can be a stand-alone server, such as ActiveMQ or Websphere MQ, or a messaging implementation that is part of an application server like the JBoss Application Server. The latter is often the easiest solution, as long as sender and receiver both run inside the same application server.
Destinations are the things you send messages to and receive messages from. There are two kinds of destination defined in the JMS standard:
- Queues allow only one receiver per message. If the message is read and acknowledged, it will be removed from the queue.
- Topics allow several subscribers to receive messages. A message stays in the Topic until all subscribers read and acknowledged it (more).
Both topics and queues have in common that there can be several senders, and that messages will always be received in the order they have been sent.
Sending and Receiving Messages (Standalone Client)
ConnectionFactory factory =
new ActiveMQConnectionFactory("tcp://localhost:61616"); // ActiveMQ-specific (more)
Connection con = factory.createConnection();
try {
Session session =
con.createSession(false, Session.AUTO_ACKNOWLEDGE); // non-transacted session (more)
Queue queue = session.createQueue("test.queue"); // only specifies queue name (more)
MessageProducer producer = session.createProducer(queue);
Message msg = session.createTextMessage("hello queue"); // text message (more)
producer.send(msg);
}
finally {
con.close(); // free all resources (more)
}
ConnectionFactory factory =
new ActiveMQConnectionFactory("tcp://localhost:61616"); // ActiveMQ-specific (more)
Connection con = factory.createConnection();
try {
Session session =
con.createSession(false, Session.AUTO_ACKNOWLEDGE); // non-transacted session (more)
Queue queue = session.createQueue("test.queue"); // only specifies queue name (more)
MessageConsumer consumer = session.createConsumer(queue);
con.start(); // start the connection (more)
while (true) { // run forever
Message msg = consumer.receive(); // blocking! (more)
if (! (msg instanceof TextMessage))
throw new RuntimeException("Expected a TextMessage");
TextMessage tm = (TextMessage) msg;
System.out.println(tm.getText()); // print message content
}
}
finally {
con.close(); // free all resources (more)
}
ConnectionFactory factory =
new ActiveMQConnectionFactory("tcp://localhost:61616"); // ActiveMQ-specific (more)
Connection con = factory.createConnection();
try {
Session session =
con.createSession(false, Session.AUTO_ACKNOWLEDGE); // non-transacted session (more)
Queue queue = session.createQueue("test.queue"); // only specifies queue name (more)
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message msg) {
try {
if (! (msg instanceof TextMessage))
throw new RuntimeException("no text message");
TextMessage tm = (TextMessage) msg;
System.out.println(tm.getText()); // print message
}
catch (JMSException e) {
System.err.println("Error reading message");
}
}
});
con.start(); // start the connection (more)
Thread.sleep(60 * 1000); // receive messages for 60s (more)
finally {
con.close(); // free all resources (more)
}
Sending and Receiving Messages in EJB 3 Applications
// MySenderLocal.java: the interface of the sending bean
@Local
public interface MySenderLocal {
void sendMessage(String txt) throws JMSException;
}
// MySenderLocal.java: implementation of MySenderLocal
@Stateless
public class MySender implements MySenderLocal {
@Resource(mappedName="/ConnectionFactory") // inject ConnectionFactory (more)
private ConnectionFactory factory;
@Resource(mappedName="/queue/JarfillerQueue") // inject Queue (more)
private Queue target;
// Sends the given string as text message:
public void sendMessage(String txt) throws JMSException {
Connection con = factory.createConnection();
try {
Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(target);
producer.send(session.createTextMessage(txt));
}
finally {
con.close();
}
}
}
@MessageDriven( // Message-driven bean (MDB)
activationConfig = { @ActivationConfigProperty( // activation configuration (more)
propertyName = "destinationType",
propertyValue = "javax.jms.Queue"
),
@ActivationConfigProperty(
propertyName = "destination",
propertyValue = "/queue/JarfillerQueue" // destination's JNDI name
) })
public class MyMDB implements MessageListener {
public void onMessage(Message message) { // process message here
TextMessage textMsg = (TextMessage) message;
try {
System.out.println("Got message: " + textMsg.getText());
}
catch (JMSException e) {
System.out.println("Error retrieving message content");
}
}
}
Using Topics
Sending messages to a Topic is identical to sending them to a Queue. Just obtain a Topic instead of the Queue. If you are writing a stand-alone client, use createTopic instead of createQueue to get the Topic instance.
Connection con = ...;
Session session = ...;
Topic topic = ...;
MessageConsumer consumer = session.createConsumer(topic); // subscribe (more)
con.start(); // start the connection
Now you have a non-durable subscription. You can either poll for or listen to messages, as you would on a queue.
In order to end the non-durable subscription, close either the MessageConsumer or the whole Connection:
consumer.close(); // unsubscribe (nondurable only)
Connection con = ...;
Session session = ...;
Topic topic = ...;
con.setClientID("JmsTest"); // required! (more)
MessageConsumer consumer = session.createDurableSubscriber(queue, "s1"); // subscribe (more)
con.start();
Now you have a durable subscription. If you used the same client ID and subscription name again, you connect to an existing subscription. Otherwise you create a new one.
The subscription exists until you explicitly unsubscribe by calling the unsubscribe method. Closing the MessageConsumer does not end the durable subscription.
A durable subscription must be explicitly unsubscribed (non-durable subscriptions end automatically after closing).
Session session = ...;
session.unsubscribe("s1"); // no open MessageConsumer allowed here (more)
Message Types
The TextMessage contains a string. If you want to communicate with non-JMS systems, text messages are often the right choice (with XML in a BytesMessage being the main alternative), as they are supported almost everywhere and are not JMS- or Java-specific.
// Create message
Session session = ...;
TextMessage msg = session.createTextMessage("some text here");
// Read message
String text = msg.getText();
The MapMessage allows you to store name/value pairs. The names are always strings, and the values can be either Java primitives, strings or byte arrays. The pairs do not have an order.
// Create message
Session session = ...;
MapMessage msg = session.createMapMessage();
msg.setString("title", "Thriller");
msg.setInt("releaseYear", 1982);
msg.setDouble("millionsSold", 110.3);
// Read message
int releaseYear = msg.getInt("releaseYear"); // can be read in any order
String title = msg.getString("title");
double millionsSold = msg.getDouble("millionsSold");
The StreamMessage allows you to store a stream of Java number and string values. When received, the values must be read in exactly the same order as they have been written.
// Create message
Session session = ...;
StreamMessage msg = session.createStreamMessage();
msg.writeString("Thriller");
msg.writeInt(1982);
msg.writeDouble(110.3);
// Read message
String title = msg.readString(); // must be read in the same order as written
int releaseYear = msg.readInt();
double millionsSold = msg.readDouble();
The ObjectMessage allows you to store any serializable Java object.
// Create message
Session session = ...;
Date obj = new Date(); // the object to send
ObjectMessage msg = session.createObjectMessage(obj);
// Read message
Date d = (Date) msg.getObject();
The BytesMessage is the raw equivalent of the StreamMessage. It allows you to write the bytes directly in a server-specific format. Its mostly useful for transmitting byte arrays.
// Create message
Session session = ...;
byte[] blob = ...; // any binary data
BytesMessage msg = session.createBytesMessage();
msg.writeBytes(blob);
// Read message
byte[] newBlob = new byte[(int) msg.getBodyLength()];
msg.readBytes();
It is also possible to create untyped messages without content (they may still contain header data and properties). This may be useful, for example, if the message should only trigger an action and does not need any additional data.
Session session = ...;
Message msg = session.createMessage();
How to...
And now...?
This tutorial intends to give you a quick start on using JMS, and maybe give answers for some questions that are hard to figure out. But now you are on you own. If you need to find out more, try to find it in the API Documents. For a more thorough understanding of JMS you should take the time to read the JMS Specification.