ActiveMQ Introduction

ActiveMQ is a messaging bus and helps in building Event based systems.It supports JMS which is a JavaEE based feature. In this tutorial, we will look into how to install and start a ActiveMQ server and a small program to create topic and consume topic over the message bus. We will also look into the ActiveMQ web front end where we can see what is happening in the message bus. This tutorial is based on windows but pretty much similar steps can be followed for Linux.

Installation of ActiveMQ

• Download the latest version from http://activemq.apache.org/download-archives.html . This post is done with version 5.9.1
• Download the zip file and extract it to some location.
• In the extracted folder you would see a bin directory. Inside that will be batch file activemq
• Run the activemq batch script
• A cmd window would appear and if everything is fine you should be able to access the ActiveMQ management console at http://localhost:8161/ 

ActiveMQ is a messaging bus so it works on the notion of consumer and producer. A producer produces the message and consumer consumes it. Producers can produce message both on Queues and Topics. Let's write a TopicProducer and TopicConsumer to see how we can interact with ActiveMQ.

Maven dependencies

To connect to activeMQ server you will need client libraries. The corresponding dependency for that is as follows. In your case, you can look into maven repository and take the latest one.

<dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-core</artifactId>

<version>5.7.0</version>

</dependency>

Writing a Producer

A producer is responsible for producing the message. A representative producer looks as follows

public class TopicProducer implements Runnable {

//Connection Factory which will help in connecting to ActiveMQ serer

ActiveMQConnectionFactory connectionFactory = null;

public TopicProducer(ActiveMQConnectionFactory connectionFactory) {

this.connectionFactory = connectionFactory;

}

@Override

public void run() {

try {

// First create a connection

Connection connection = 

connectionFactory.createConnection();

connection.start();

// Now create a Session

Session session = connection.createSession(false,

Session.AUTO_ACKNOWLEDGE);

// Let's create a topic. If the topic exist, 

//it will return that

Destination destination = session.createTopic("CLIMATE");

// Create a MessageProducer from 

//the Session to the Topic or Queue

MessageProducer producer = 

session.createProducer(destination);

producer.setDeliveryMode(DeliveryMode.PERSISTENT);

// Create a messages for the current climate

String text = "Today is Hot";

TextMessage message = session.createTextMessage(text);

// Send the message to topic

producer.send(message);

// Do the cleanup

session.close();

connection.close();

} catch (JMSException jmse) {

         System.out.println("Exception: " + jmse.getMessage());

}

}

}

Writing a Consumer

public class TopicConsumer implements Runnable {

ActiveMQConnectionFactory connectionFactory = null;

public TopicConsumer(ActiveMQConnectionFactory connectionFactory) {

this.connectionFactory = connectionFactory;

}

@Override

public void run() {

try {

// First create a connection

Connection connection = 

connectionFactory.createConnection();

connection.start();

// Now create a Session

Session session = connection.createSession(false,

Session.AUTO_ACKNOWLEDGE);

// Let's create a topic. If the topic exist, 

//it will return that

Destination topicDestination = 

session.createTopic("CLIMATE");

// Create a MessageProducer from the Session 

//to the Topic or Queue

MessageConsumer messageConsumer =

session.createConsumer(topicDestination);

//Get the message

Message message = messageConsumer.receive();

TextMessage textMessage = (TextMessage)message;

System.out.println(textMessage.getText());

// Do the cleanup

session.close();

connection.close();

} catch (JMSException jmse) {

System.out.println("Exception: " + jmse.getMessage());

}

}

}

Writing the main program

//Create the connection factory

ActiveMQConnectionFactory connectionFactory = 

new ActiveMQConnectionFactory("tcp://localhost:61616");

//Create the consumer. It will wait to listen to the Topic

Thread topicConsumerThread = 

new Thread(new TopicConsumer(connectionFactory));

topicConsumerThread.start();

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

//Create a message. As soon as the message is published on the Topic,

//it will be consumed by the consumer

Thread topicProducerThread = 

new Thread(new TopicProducer(connectionFactory));

topicProducerThread.start();

Run the main program and you will see the consumer consuming the message and printing it on console.

Code at GitHub