Enterprise Java Development@TOPIC@

• Prev
• Next

Chapter 123. JMS API

123.1. JMS API Classes

123.2. ConnectionFactory

123.3. JMSContext

123.4. Destinations

123.4.1. JNDI Lookup

123.4.2. Injected Destination

123.5. Send Message

123.6. Receive Message

123.6.1. Poll from JMSConsumer

123.6.2. JMSConsumer Listener

123.6.3. Message Driven Bean (MDB)

123.7. Message Acknowledgement

123.7.1. Message Acknowledgement: AUTO_ACKNOWLEDGE

123.7.2. Message Acknowledgement: DUPS_OK_ACKNOWLEDGE

123.7.3. Message Acknowledgement: CLIENT_ACKNOWLEDGE

123.7.4. Message Acknowledgement: SESSION_TRANSACTED

123.8. JMS API Summary

123.1. JMS API Classes

• JMS 2.0 "Ease of Use"

  • Collapsed Connection and Session functionality into JMSContext

  • Created easier to manage/use JMSProducer and JMSConsumer

123.2. ConnectionFactory

• Encapsulates a set of properties usually assigned by administrator

  • clientId - optional unique identity required for durable subscriptions

  • listen address

• commonly obtained thru JNDI lookup

  import javax.naming.InitialContext;
  
  import javax.jms.ConnectionFactory;
  

  InitialContext jndi = new InitialContext();    
  
  ConnectionFactory connFactory=(ConnectionFactory)jndi.lookup("jms/RemoteConnectionFactory");
  

123.3. JMSContext

• Client's primary interface to JMS provider

  • Combines Connection and Session functionality

  • Allocates resources outside of client JVM

• Created from ConnectionFactory

  import javax.jms.JMSContext;
  

  JMSContext context = connFactory.createContext()
  

• Must be closed (except when injected)

• Supports JavaSE 7 AutoCloseable

  try (JMSContext context = connFactory.createContext()) {
  
    ...
  
  }
  

• Authentication can be performed when created from JavaSE client

  JMSContext context = connFactory.createContext(user, password)
  

• Can be injected when running within JavaEE container

  import javax.jms.JMSConnectionFactory;
  

  @Inject @JMSConnectionFactory("java:/JmsXA")
  
  private JMSContext jmsContext;
  

• One thread per JMSContext

  • application may create child JMSContext for multi-threaded applications

  try (JMSContext context=createContext();
  
       JMSContext context2=context.createContext(JMSContext.CLIENT_ACKNOWLEDGE)) {
  
     ...
  
  }
  

• Supports an ExceptionListener

123.4. Destinations

• Works the same for Topic and Queue

• JNDI lookups supported for JavaSE clients

• Injection supported for JavaEE components

123.4.1. JNDI Lookup

import javax.jms.Destination;

Destination destination = (Destination)jndi.lookup("jms/queue/ejava/examples/jmsMechanics/queue1")

123.4.2. Injected Destination

import javax.jms.Topic;

@Resource(lookup="java:/jms/topic/ejava/examples/asyncMarket/topic1", type=Topic.class)

private Destination sellTopic;

123.5. Send Message

1. Obtain JMSContext from ConnectionFactory (JavaSE) or injection (JavaEE container)

2. Obtain Destination from JNDI lookup (JavaSE) or injection (JavaEE container)

3. Create Message

   • Message - all properties and no payload

     • Messages have pre-defined and user-defined properties

     • Message properties are subject to Subscription Selectors in broker

     • Messages are not closed

   • MapMessage - Map payload of properties

     • Additional properties opaque to broker

     • Properties required to be of "java.lang" types - no custom types

   • TextMessage - textual payload

     • Ideal for XML and JSON-based DTO payloads

   • BytesMessage - binary blob payload

   • ObjectMessage - Java serialized object payload

     • Requires Java-based receiver with DTO Classes

   • StreamMessage - sequential data payload

   import javax.jms.MapMessage;
   

   MapMessage message = jmsContext.createMapMessage();
   

4. Set Message Properties (optional)

   • Message property accessesors throw JMSException checked exceptions

   import javax.jms.JMSException;
   

   message.setJMSType("saleUpdate"); //JMSType is a pre-defined property
   
   message.setStringProperty("awayClub", ...);//"awayClub" is example of user-defined String property
   
   message.setIntProperty("awayTeamId", ...); //"awayTeamId" is example of user-defined int property
   

5. Set Message Payload

   //this example is a MapMessage - each Message type has specific interface
   
   message.setLong("id", item.getId());
   
   message.setString("name", item.getName());
   
   message.setString("seller", item.getOwner().getUserId());
   
   message.setLong("startDate", item.getStartDate().getTime());
   
   message.setLong("endDate", item.getEndDate().getTime());
   
   message.setDouble("minBid", item.getMinBid());
   
   message.setDouble("bids", item.getBids().size());
   
   message.setDouble("highestBid", 
   
           (item.getHighestBid() == null ? 0.00 : item.getHighestBid().getAmount()));            
   

6. Create JMSProducer

   • JMSProducers are not closed - different from JMS 1.1 Producer

   import javax.jms.Producer;
   

   JMSProducer producer = context.createProducer();
   

7. Set JMSProducer options

   • Time to live - number of millisecs a sent message will live in the broker waiting to be delivered

   • Priority - delivery priority of messages given to broker

   • Delivery Mode - level of guarantee required when completing a send() to the broker

   • Delivery Delay - minimum number of millsecs before broker will issue to consumer

   //JMSProducer supports method chaining
   
   producer.setPriority(Message.DEFAULT_PRIORITY)
   
           .setTimeToLive(Message.DEFAULT_TIME_TO_LIVE)
   
           .setDeliveryMode(Message.DEFAULT_DELIVERY_MODE)
   
           .setDeliveryDelay(Message.DEFAULT_DELIVERY_DELAY);
   

8. Send Message to Destination

   producer.send(sellTopic, message);
   

123.6. Receive Message

123.6.1. Poll from JMSConsumer

1. Obtain JMSContext from ConnectionFactory (JavaSE) or injection (JavaEE container)

2. Obtain Destination from JNDI lookup (JavaSE) or injection (JavaEE container)

3. Create a JMSConsumer for a specific destination

   • JMSConsumer must be closed and is AutoCloseable

   import javax.jms.JMSConsumer;
   

   try (JMSConsumer syncConsumer = context.createConsumer(destination);
   
        JMSConsumer asyncConsumer = context2.createConsumer(destination)) {
   
   }
   

4. Call receive() to obtain the next message

   • receive() - perminently wait for next Message

   • receiveNoWait() - immediately return with next Message or null if none waiting

   • receive(long timeout) - wait up to timeout msecs before returning null if no Message

   Message message=consumer.receiveNoWait();
   

5. Get Message Properties (optional)

   String level = message.getStringProperty("level");//"level" is user-defined property            
   
   logger.debug("receive ({}, mode={}, pri={}{}):{}",++count,
   
           message.getJMSDeliveryMode(),       //"JMSDeliveryMode" is pre-defined property
   
           message.getJMSPriority(),           //"JMSPriority" is pre-defined property
   
           (level==null?"":", level="+level),
   
           message.getJMSMessageID());         //"JMSMessageID" is pre-defined property
   

6. Get Message Payload

   TextMessage m1 = ...
   
   TextMessage text = m1.request.getText();
   
   
   ObjectMessage m2 = ...
   
   XxxDTO dto = (XxxDTO)m2.getObject();
   

7. Alternately, call JMS 2.0 receiveBody(T) to get just the payload

   • message.receiveBody(Class<T> c)

   • message.receiveBodyNoWait(Class<T> c)

   • message.receiveBody(Class<T> c, long delay)

123.6.2. JMSConsumer Listener

1. Obtain JMSContext from ConnectionFactory (JavaSE)

2. Obtain Destination from JNDI lookup (JavaSE)

3. Implement javax.jms.MessageListener

   public class AsyncClient implements MessageListener ... {
   
       public void onMessage(Message message) {
   
         ...
   
       }
   
   }
   

4. Create a JMSConsumer

   • JMSConsumer must be closed and implements AutoCloseable

   try (JMSConsumer syncConsumer = context.createConsumer(destination);
   
        JMSConsumer asyncConsumer = context2.createConsumer(destination)) {
   
      ...
   
   }
   

5. Register MessageListener with JMSConsumer

   //create a client to asynchronous receive messages through onMessage() callbacks
   
   AsyncClient asyncClient = new AsyncClient();
   
   asyncConsumer.setMessageListener(asyncClient);
   

6. Process Messages as they are received through onMessage() callback

123.6.3. Message Driven Bean (MDB)

1. Implement a javax.jms.MessageListener

   public class BuyerMDB implements MessageListener {
   
       public void onMessage(Message message) {
   

2. Annotate the class as @MessageDriven

   • Fill in @MessageDriven @ActivationConfigProperty values

     • destinationLookup - JNDI name for destination

     • destinationType - javax.jms.Queue or javax.jms.Topic

     • messageSelector - SQL-like property selector

     • acknowledgeMode - when to 100% accept message from broker

     • subscriptionDurability - durable or transient Topic subscription

     • clientId - unique JMSContext/Connection ID for durable subscription

     • subscriptionName - required for durable subscription

     • connectionFactoryLookup - JNDI name for ConnectionFactory

   
   
   @MessageDriven(activationConfig={
   
           @ActivationConfigProperty(
   
                   propertyName="destinationType",
   
                   propertyValue="javax.jms.Topic"),            
   
           @ActivationConfigProperty(
   
                   propertyName="destination",
   
                   propertyValue="java:/jms/topic/ejava/examples/asyncMarket/topic1"),            
   
           @ActivationConfigProperty(
   
                   propertyName="messageSelector",
   
                   propertyValue="JMSType in ('forSale', 'saleUpdate')"),
   
           @ActivationConfigProperty(
   
                   propertyName="acknowledgeMode",
   
                   propertyValue="Auto-acknowledge")            
   
   })
   
   public class BuyerMDB implements MessageListener {
   
       @PermitAll
   
       public void onMessage(Message message) {
   

3. Grant container permission to invoke onMessage()

   • Annotate class or onMessage() method with @PermitAll

   
   
   @MessageDriven(...)
   
   public class BuyerMDB implements MessageListener {
   
       @PermitAll
   
       public void onMessage(Message message) {
   

4. Process Messages as they are received through onMessage() callback

123.7. Message Acknowledgement

• Broker retains consumed until acknowledged

• Smaller windows (ack per Message) slower but can keep client and broker fully synchronized

• Larger windows (ack per N Messages) faster but can produce duplicate messages

123.7.1. Message Acknowledgement: AUTO_ACKNOWLEDGE

try (JMSContext context = connFactory.createContext(JMSContext.AUTO_ACKNOWLEDGE)) {

    ...

    Message message = ...

}

• Message automatically acknowledged by session when client synchronously receives (receive()) or asynchronously processes (onMessage) message

• Can be used with MessageDriven EJBs (MDBs)

  @MessageDriven(activationConfig={
  
          @ActivationConfigProperty(
  
                  propertyName="acknowledgeMode",
  
                  propertyValue="Auto-acknowledge")            
  
  })
  
  public class BuyerMDB implements MessageListener {
  
      @PermitAll
  
      public void onMessage(Message message) {
  
          ...
  
      }
  
  }
  

123.7.2. Message Acknowledgement: DUPS_OK_ACKNOWLEDGE

try (JMSContext context = connFactory.createContext(JMSContext.DUPS_OK_ACKNOWLEDGE)) {

    ...

    Message message = ...

}

• Similar to AUTO_ACKNOWLEDGE

• Session lazily acknowledges messages

• Can result in duplicate messages

• Can also be used with MessageDriven EJBs (MDBs)

  @MessageDriven(activationConfig={
  
          @ActivationConfigProperty(
  
                  propertyName="acknowledgeMode",
  
                  propertyValue="Dups-ok-acknowledge")            
  
  })
  
  public class BuyerMDB implements MessageListener {
  
      @PermitAll
  
      public void onMessage(Message message) {
  
          ...
  
      }
  
  }
  

123.7.3. Message Acknowledgement: CLIENT_ACKNOWLEDGE

try (JMSContext context = connFactory.createContext(JMSContext.CLIENT_ACKNOWLEDGE)) {

    ...

    Message message = ...

    message.acknowledge();

}

• Messages are manually acknowledged

• Any acknowledged message acknowledges all prior messages consumed

123.7.4. Message Acknowledgement: SESSION_TRANSACTED

try (JMSContext context = connFactory.createContext(JMSContext.SESSION_TRANSACTED)) {

    ...

    Message message = ...

    context.commit();

}

• Send or receive Message(s) within a local transaction

• Changes are not reflected in broker until commit() and undone on call to rollback()

• Useful for JavaSE clients to receive and fully process message

• Can only be used with MDBs using Bean-Managed Transactions

123.8. JMS API Summary

• ConnectionFactory can contain client-specific properties (e.g., clientId)

• Flows begin with JMSContext and Destination

• JMSProducer required to send()

• JMSConsumer required to receive() or onMessage()

• PrevChapter 122. JMS
• Top of page
• Front page
• NextChapter 124. JMS Maven Aspects