JMS Sample

Java Message Service

JMS APIs provides the way to send, receive messages in enterprises asynchronous way. There is no need of wait for the message to get processed in server and continue the business in client. Just post, and let it get processed. These APIs are bundles under the package javax.jms and JAVA Documentation found at JMS APIs

JMS comes with flavours of messaging - Queue(point-to-point destination) and Topic(publish/subscribe model). Best sample usecase would be, Queue will be useful to communicate any one consumer has to act on request. In cluster environment, Topic helps to send broadcast message to all the nodes.

• Message types can be byte, text or object messages. JMS provides API to construct message based on the message - javax.jms.TextMessage, javax.jms.ByteMessage, javax.jms.StreamMessage and javax.jms.ObjectMessage.
• javax.jms.ConnectionFactory helps to define connection configuration properties such as Transaction, LoadBalancing and etc.,. Using ConnectionFactory, Connection get created. Using this javax.jms.Connection, javax.jms.Session is created to post/receive message.
• Queue/Topic helps to locate or point the destination where the message has to go and/or pulled from.
• javax.jms.QueueSession exposes APIs to create sender, receiver, browser and Queue.
  QueueBrowser used to read message from Queue without deleting from Queue.
• javax.jms.TopicSession exposes APIs to create publisher, consumer, and Topic.

Here is simple example code which helps to post/receive text message from weblogic server.

import java.util.Hashtable;

import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class QSendTest {
 public static void main(String[] args) 
 throws Exception {
  QueueConnectionFactory qconFactory = null;
  QueueConnection qcon = null;
  QueueSession qsession = null;
  Queue queue = null;
  TextMessage msg = null;

  InitialContext ctx = getInitialContext();
  qconFactory = (QueueConnectionFactory) 
  ctx.lookup("jms/MyQueueFactory");
  qcon = qconFactory.createQueueConnection();
  qsession = qcon.createQueueSession(false, 
   QueueSession.AUTO_ACKNOWLEDGE);

  queue = (Queue) ctx.lookup("jms/MyQueue");
  
//Posting message to Queue
  QueueSender qsender = qsession.createSender(queue);
  msg = qsession.createTextMessage();
  msg.setStringProperty("MSG_ID", "12345678947");
  msg.setText("Hello at " + System.currentTimeMillis());
  qsender.send(msg);

  System.out.println("Sent");
  qsender.close();

//To receive message from Queue
  qcon.start();
  QueueReceiver qReceiver = qsession.createReceiver(queue);
  while (true) {
   TextMessage txt = (TextMessage)
   qReceiver.receiveNoWait();
   if (txt == null)
    break;
   System.out.println(txt.getText() 
   + "with header MSG_ID="
   + txt.getStringProperty("MSG_ID"));
  }

  qReceiver.close();
  qsession.close();
  qcon.close();
 }

 private static InitialContext getInitialContext()
 throws NamingException {
  Hashtable<String, String> env = new Hashtable<String, String>();
  env.put(Context.INITIAL_CONTEXT_FACTORY,
    "weblogic.jndi.WLInitialContextFactory");
  env.put(Context.PROVIDER_URL, "t3://localhost:7001/");
  return new InitialContext(env);
 }
}

In weblogic, javax.jms_1.1.1.jar has JMS APIs and its implementaions. Before running this program, we have set up JMS configuration in Server. Steps are given below

Create JMS Server

1. Login weblogic console, select services from Domain structure tab.
2. Select JMS Servers
3. In JMS Servers tab, click New
4. Type MyJMsServer and select AdminServer
5. click on finish

Create JMS Module

1. Login weblogic console, select services from Domain structure tab.
2. Select JMS Modules
3. In JMS Modules tab, click New
4. Type MyJMsModule and select AdminServer
5. click on finish

Create Subdeployments

1. Login weblogic console, select services from Domain structure tab.
2. Select MyJMsModule and select subdeployments
li>In Summary of Resources panel, click New
3. Type MyQueue and select MyJMsServer
4. click on finish

Create JMS Connection Factory

1. Login weblogic console, select services from Domain structure tab.
2. Select JMS Modules and select MyJMsModule
3. In Summary of Resources panel, click New
4. Select Connection Factory and say Next
5. Type Name as MyQueueFactory and JNDI Name as jms/MyQueueFactory
6. Click Next and Finish

Create JMS Queue

1. Login weblogic console, select services from Domain structure tab.
2. Select JMS Modules and select MyJMsModule
3. In Summary of Resources panel, click New
4. Select Queue and say Next
5. Type Name as MyQueue and JNDI Name as jms/MyQueue
6. Click Next and Finish

Create JMS Queue

1. Login weblogic console, select services from Domain structure tab.
2. Select JMS Modules and select MyJMsModule and MyQueue
3. In Summary of Resources panel, click New
4. Select Queue and say Next
5. Type Name as MyQueue and JNDI Name as jms/MyQueue
6. Click Next and Finish

java.util.concurrent.Semaphore

java.util.concurrent.Semaphore

If we wants to make programmatically Connection management, connection may be Database or TCP Socket or JMS or other. Instead of establishing connection every time for every SQL command, we can cache the connection for future use. Once the usage of connection usage is completed then we can take connection back to pool. For instance, We have established 20 connections, but 50 SQL commands has to be run in the given point of time. These 20 connections has to be alotted to first 20 SQL commands to run and whenever SQL command is completed then connection will be used for rest of SQL commands.

We have to maintain a mark for each connection that whether that connection is get used or not. Two methods has to be used to acquire and release the resource. Everything perfect, who will take care of filtering and ordering the thread entry, how thread state managed. For these questions Semaphore is the right answer, here is the code snippet

 class ConnectionPool {
   private static final int MAX_AVAILABLE = 100;
   private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);
 
   public Object getConnection() throws InterruptedException {
      available.acquire(); 
     return getNextAvailableItem();
   }
 
   public void releaseConnection(Object x) {
     if (markAsUnused(x))
       available.release();
   }
 
   protected Object[] connections = ... whatever kinds of connections being managed
   protected boolean[] used = new boolean[MAX_AVAILABLE];
 
   protected synchronized Object getNextAvailableItem() {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (!used[i]) {
          used[i] = true;
          return connections[i];
       }
     }
     return null; // not reached
   }
 
   protected synchronized boolean markAsUnused(Object item) {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (item == connections[i]) {
          if (used[i]) {
            used[i] = false;
            return true;
          } else
            return false;
       }
     }
     return false;
   }
 
 }