Core JMS using Apache ActiveMQ

Java Messaging Service(JMS) using ActiveMQ

In the below example, we have no need to create the queue in the ActiveMQ, after the successful execution of the producer will automatically create the Queue in the Apache ActiveMQ.

System Requirements:-

•       Eclipse Editor or any other.

•       JDK 1.5 or higher(I am using jdk 1.7.0_03)

•       Required jars(activemq-all-5.4.3.jar) as referenced library.       

•       Apache-activemq-5.4.3

Note: - Apache Active MQ Setup is required for the execution of this example. For doing the Active MQ Setup please follow the below link:-

http://www.javatechtipssharedbygaurav.com/2013/10/how-to-work-with-apache-activemq.html

Steps for creating Eclipse java project for implementing Core JMS using Apache ActiveMQ:-

• Create a java project named JMSUsingActiveMQ

• Create a package names com.gaurav.jms.activemq in the src directory

• Project Structure is as below:- 

• Create an ActiveMQMessageProducer.java in the above specified package.

package com.gaurav.jms.activemq;

import javax.jms.Connection;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQMessageProducer {

      public static void main(String args[]) {

            try {

                  // Creating a connection factory for ActiveMQ

                  ActiveMQConnectionFactory activeMQConFactory = new ActiveMQConnectionFactory(

                              "tcp://localhost:61616");

                  // Creating a connection

                  Connection con = activeMQConFactory.createConnection();

                  con.start();

                  // Creating a session;

                  Session session = con

                              .createSession(false, Session.AUTO_ACKNOWLEDGE);

                  // Creating a destination using Topic or Queue

                  Destination dest = session.createQueue("TestWelcomeActiveMQQueue");

                  // creating a MessageProducer using the session to the topic or

                  // queue.

                  MessageProducer msgProducer = session.createProducer(dest);

                  msgProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

                  // Creating a message for sending in the queue

                  String strMessage = "WELCOME GAURAV BY "

                              + Thread.currentThread().getName();

                  //Creating a text message using the session.

                  TextMessage txtMessage = session.createTextMessage(strMessage);

                  System.out.println("Sent message : " + strMessage.hashCode()

                              + " : " + Thread.currentThread().getName());

                  msgProducer.send(txtMessage);

                  // closing the resources

                  msgProducer.close();

                  session.close();

                  con.close();

            } catch (Exception e) {

                  System.out.println("Exception thrown : " + e);

                  e.printStackTrace();

            }

      }

}

/* Note:- NON_PERSISTENT means no need for database specific persistent */

•        Create an ActiveMQMessageConsumer.java in the above specified package.

  

package com.gaurav.jms.activemq;

import javax.jms.Connection;

import javax.jms.Destination;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMQMessageConsumer {

      public static void main(String args[]) {

            try {

                  // Creating a connection factory for ActiveMQ

                  ActiveMQConnectionFactory activeMQConFactory = new ActiveMQConnectionFactory(

                              "tcp://localhost:61616");

                  // Creating a connection

                  Connection con = activeMQConFactory.createConnection();

                  con.start();

                  // Creating a session;

                  Session session = con

                              .createSession(false, Session.AUTO_ACKNOWLEDGE);

                  // Creating a destination using Topic or Queue

                  Destination dest = session.createQueue("TestWelcomeActiveMQQueue");

                  // creating a MessageConsumer using the session for the topic or

                  // queue.

                  MessageConsumer msgConsumer = session.createConsumer(dest);

                  Message message = msgConsumer.receive(5000);

                  //Checking the message is an instance of textMessage or not.

                  if(message instanceof TextMessage){

                        TextMessage txtMessage = (TextMessage)message;

                        String strMessage = txtMessage.getText();

                        System.out.println("Received Message from queue is : "+strMessage);

                  }else{

                        System.out.println("Received : "+message);

                  }

                  // closing the resources

                  msgConsumer.close();

                  session.close();

                  con.close();

            } catch (Exception e) {

                  System.out.println("Exception thrown : " + e);

                  e.printStackTrace();

            }

      }

}

• Execute the ActiveMQMessageProducer.java  first and then ActiveMQMessageConsumer.java by selecting the option Run as Java Application.

Result of  ActiveMQMessageProducer.java

Message Producer output :

Sent message : 707288012 : main

Result of  ActiveMQMessageConsumer.java

Message Consumer output:

Received Message from queue is: WELCOME GAURAV BY main

Result In the ActiveMQ console

URL for opening activemq admin console: - http://localhost:8161

About Atomic Operation in Java

What is Atomic Operation in java?

Atomic means each action take place in one step without interruption or we can justify that operation is performed as a single unit of work without the possibility of interference from other operations.

An Atomic operation can't stop in the middle, either it happened completely or doesn't happen at all. No side effects of an atomic operation is visible until the action/operation is complete.

According to java language specification, it  guarantees that

·      Reads and writes are atomic for reference variables and for most primitive variables (for all primitive data types except long and double).

·      Read and write are atomic for all variable declared volatile including long and double variables.

The operation like below is not an atomic operation:-

            int i++;

The upper operation is having 3 steps to complete.

1).    Reading the current value of i;

2).    Incrementing the current value of i;

3).    Writing the modified value of i;

Example of non thread safe code in java:-

package com.gaurav.java.atomictest;

public class Counter {

      private int incrementCounter;

      /*

       * This method is not a thread safe method because ++ is not an atomic

       * operation

       */

      public int getIncrementCounter() {

            return incrementCounter++;

      }

}

In the above example, inside the Counter class the  getIncrementCounter() method is not a thread safe operation because ++(increment operator) is not atomic operation and I mentioned earlier that this can be broken down into three different steps. So if multiple threads call this getIncrementCounter() method simultaneously then each of these three operation may overlap with each other. For example while thread 1 is updating value , thread 2 reads and but still gets old value, which eventually let thread 2 override thread 1 increment and one count is lost because multiple threads are working concurrently.

Writing thread safe code in java for above scenario's

There are many ways to make the above code as thread safe in Java:

            First Approach

With the use of synchronized keyword in Java by providing locking to the getIncrementCounter() method, we can assure that only one thread can execute it at a time which removes possibility of coinciding or overlapping.

package com.gaurav.java.atomictest;

public class SynchronizeCounter {

      private int incrementCounter;

      /* This method is thread safe because of locking provided by synchronization */

      public synchronized int getIncrementCounter() {

            return incrementCounter++;

      }

}

            Second Approach

With the use of Atomic Integer which is available in java.util.concurrent.atomic package with jdk1.5 api, which helps to make this ++ operation atomic and since atomic operations are thread-safe and saves cost of external synchronization.

package com.gaurav.java.atomictest;

import java.util.concurrent.atomic.AtomicInteger;

public class AutomicIntegerCounter {

      AtomicInteger atomicCounterIncrement = new AtomicInteger(0);

      /*

       * This method is thread safe because the counter is incremented

       * automatically using the AtomicInteger class methods

       */

      public int getCountIncrementAutomatically() {

            return atomicCounterIncrement.incrementAndGet();

      }

}

Facts about thread-safety

• Immutable objects are by default thread-safe because there state can not be changed once created.

• Read only or Final variables are useful in writing of thread safe programs.

• With the use of already available thread safe classes like-StringBuffer, HashTable, Vector e.t.c

• With the use of local variables because each thread has there own copy of local variables.

• By minimizing sharing of objects between multiple thread we can also avaiod the issue of thread safely.

• Volatile keyword in Java can also be used to instruct thread not to cache variables and read from main memory and can also instruct JVM not to reorder or optimize code from threading perspective.