SpringJMS - Step by Step Guide to create Consumer (Super Simple)

Step 1 - You will need same Jar files which we discussed in Producer Example

Spring Core
Spring Context
Spring JMS
activemq-all
commons-logging
slf4j

Step 2: Create applicationContext.xml in classpath

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:p="http://www.springframework.org/schema/p"

       xmlns:context="http://www.springframework.org/schema/context"

       xmlns:jms="http://www.springframework.org/schema/jms"

       xmlns:amq="http://activemq.apache.org/schema/core"

       xsi:schemaLocation="http://www.springframework.org/schema/beans 

                           http://www.springframework.org/schema/beans/spring-beans.xsd

                           http://www.springframework.org/schema/context 

                           http://www.springframework.org/schema/context/spring-context.xsd

                           http://www.springframework.org/schema/jms 

                           http://www.springframework.org/schema/jms/spring-jms.xsd

                           http://activemq.apache.org/schema/core 

                           http://activemq.apache.org/schema/core/activemq-core.xsd">

<context:component-scan base-package="com.itech" />

<context:annotation-config/>

<!-- JMS ConnectionFactory to use, configuring the embedded broker using XML -->

<bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

    <!-- brokerURL, You may have different IP or port -->

    <constructor-arg name="brokerURL" value="vm://localhost:61616" />

  </bean>  

<!-- Pooled Spring connection factory -->

 <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">

    <constructor-arg ref="activemqConnectionFactory" />

 </bean>

<!--  Queue Destination  -->

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

    <!-- name of the queue -->

    <constructor-arg index="0" value="THIS.IS.TEST.QUEUE" />

  </bean>

 

  <!-- Listeners   -->

<jms:listener-container connection-factory="connectionFactory"> 

               <jms:listener destination="THIS.IS.TEST.QUEUE" ref="messageListenerTest" /> 

 </jms:listener-container>

</beans>

Step 3:

You need a Consumer Class which implements MessageListener interface:

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

import org.springframework.stereotype.Component;

@Component("messageListenerTest")

public class MessageListenerTest implements MessageListener {

@Override

public void onMessage(Message message) {

TextMessage textMessage = (TextMessage) message;

try {

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

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

That's it, Now try sending a message with Producer example, and message will be listened by this MessageListener implementation. 

ActiveMQ Composite Destinations

Composite queue allows one-to-many relationship. Send a message to One Queue (A) and it will forward the message to Queue B, Queue C and Topic D.

To map Composite destination you need to add following section in virtualDestinations tag.

 <destinationInterceptors>

    <virtualDestinationInterceptor>

        <virtualDestinations>

<compositeQueue name="TEST.QUEUE.A">

   <forwardTo>

<queue physicalName="TEST.QUEUE.B" />

<queue physicalName="TEST.QUEUE.C" />

<queue physicalName="TEST.TOPIC.D" />

   </forwardTo>

</compositeQueue>

        </virtualDestinations>

    </virtualDestinationInterceptor>

  </destinationInterceptors>

By default, subscribers cannot consume messages directly from a composite queue or topic - it is a logical construct only. Given the configuration above, subscribers can only consume messages from TEST.QUEUE.B, TEST.QUEUE.C and TEST.TOPIC.D; but NOT TEST.QUEUE.A.

Using filtered destinations

You may wish to create a virtual destination which forwards messages to multiple destinations but applying a selector first to decide if the message really does have to go to a particular destination.

 <destinationInterceptors>

    <virtualDestinationInterceptor>

        <virtualDestinations>

<compositeQueue name="TEST.QUEUE.A">

   <forwardTo>

<filteredDestination selector="criteria = ‘B’” queue="TEST.QUEUE.B" />

<filteredDestination selector="criteria = ‘C’” queue="TEST.QUEUE.C" />

<filteredDestination selector="criteria = ‘D’” queue="TEST.TOPIC.D" />

   </forwardTo>

</compositeQueue>

        </virtualDestinations>

    </virtualDestinationInterceptor>

  </destinationInterceptors>

In the case above, message from TEST.QUEUE.A will be forwarded to filtered destinations with selector criteria. Result is same as having selector based consumer on TEST.QUEUE.B

Important Aspects of Composite Destinations:

-          Listener Queues (under forwardTo Tag) will not be created automatically. If queues are not present, message will be LOST.

-          If you want to store message in composite queue, make forwardOnly flag as ‘false’. It’s true by default.

<compositeQueue name="TEST.QUEUE.A" forwardOnly="false">   

      <forwardTo>

<queue physicalName="TEST.QUEUE.B" />

   </forwardTo>

</compositeQueue>

-          Composite queue can send the message to Virtual Topic, which will be further consumed by multiple Virtual Destinations.

-          Virtual Topic Selector Aware Consumers will not pull the message if consumer is not active; On the contrary, Filtered Composite destinations pull the message despite of consumer’s active status. This is very helpful in real world scenarios.

-          Important: Composite Destinations Supports wildcards. This is also valid for Filtered Destinations.

<destinationInterceptors>

   <virtualDestinationInterceptor>

      <virtualDestinations>

<compositeQueue name="TEST.QUEUE.A">

<forwardTo>

   <filteredDestination selector="criteria = ‘B’” queue="TEST.B.>."/>

   <queue physicalName="COMPOSITE.WILD.>."/>

</forwardTo>

</compositeQueue>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

In the settings above, message will be forwarded to all the queues starting with “COMPOSITE.WILD”. Also, the message with filtered destinations will be forwarded to all the queues starting with ‘TEST.B’.

WTX with ActiveMQ Using JMS Adapter

WebSphere Transformation Extender's interaction with Activemq. Not hard but little tricks are important. As usual, if you google this solution, you will not find a single site or blog stating exact steps. So, here it is. Post your comments for questions.

Step 1: Setting jndi.propertise

Place provided “jndi.properties” file in <WTX_INSTALL>/java/lib directory or defined CLASSPATH. Sample jndi.properties can be downloaded from here.

http://activemq.apache.org/jndi-support.html

Step 2: Include Jars

Include following Jars in local accessible folder.

activemq-all-5.5.1-fuse-07-11.jar

geronimo-jms_1.1_spec-1.1.1.jar

log4j-1.2.16.jar

slf4j-api-1.6.4.jar

slf4j-log4j12-1.6.4.jar

spring-jms-3.0.6.RELEASE.jar

Step 3: Add External Jars in dtx.ini

Include above mentioned jars in External jar section dtx.ini file. (replace home/aad/tmp with your folder path from root)

jar1=/home/aad/tmp/activemq-all-5.5.1-fuse-07-11.jar

jar2=/ home / aad /tmp/geronimo-jms_1.1_spec-1.1.1.jar

jar3=/ home / aad /tmp/log4j-1.2.16.jar

jar4=/ home / aad /tmp/slf4j-api-1.6.4.jar

jar5=/ home / aad /tmp/slf4j-log4j12-1.6.4.jar

jar6=/ home / aad /tmp/spring-jms-3.0.6.RELEASE.jar

Step 4: Command

-T -HDR -ICTXFURL tcp://<brokerURL>:<port>

 -ICTXF org.apache.activemq.jndi.ActiveMQInitialContextFactory

-CFN "connectionFactory"

-QN " testQueue"

Where

connectionFactory – JNDI Connection Factory Name

testQueue – JNDI Queue Name

ActiveMQ Mirror Queue: Working Example.

If you reached here, you might have already checked few very confusing paragraphs about Mirror queues in ActiveMQ.

Mirror Queues are disabled by default. To enable them, you have to following settings in activemq.xml

<destinationInterceptors>

<mirroredQueue copyMessage = "true" postfix="" prefix="MIRROR."/>

                <virtualDestinationInterceptor>

                                <virtualDestinations>

                                                <virtualTopic name="MIRROR.TEST.>"  prefix="MIR.>." selectorAware="false"/>

                                </virtualDestinations>

                </virtualDestinationInterceptor>

</destinationInterceptors>

               

Out of above: “<mirroredQueue copyMessage = "true" postfix="" prefix="MIRROR."/>” is actually responsible for enabling Mirror Queues.

How ActiveMQ Handles Mirror Queues:

1.       When you enable Mirror Queues, with say prefix as “MIRROR.”, for every queue in the broker one Topic will be created with MIRROR. Prefix.

2.       Now to receive a message from this Topic, Virtual Destinations can be used. More about Virtual Destination can be found here.

Working Example:

                Let’s consider above configuration for prefix for Mirror queues and Virtual Destinations.

1.       Create a queue: TEST.QUEUE

2.       With above settings, Mirror Topic is automatically created as : MIRROR.TEST.QUEUE

3.       Create Virtual Topic Consumer : MIR.COPY.MIRROR.TEST.QUEUE

a.       MIR – Defined Virtual Topic Interceptor

b.      COPY – Name of the Consumer

c.       MIRROR.TEST.QUEUE – is a Topic Name

4.       Send a test message to TEST.QUEUE, same message will appear in MIR.COPY.MIRROR.TEST.QUEUE.

This is complete setting of Mirror Queues.

Problems:

1.       Mirror Topics will be created for Every Queue in the broker

2.       As broker has to track each message from the queue and copy it to topic, further to virtual destination, broker may face performance problem depending on load.

ActiveMQ Message Sequencing : Analysis Part I

Message Sequencing POC:

In most of the corporate production environments, messages are not sent and received from the same broker. Message might end up from multiple hops and clusters within infrastructure before it actually hits the business consumer. I did a small analysis with ActiveMQ out of the box configuration to use sequencing. Below are the results.

Test Case 1:

Messages are sent to queue / Topic and received in same broker (even from Virtual Destination queue). Single producer and exclusive consumer are used.

Result:

1.       Messages are always received in Sequence in normal running consumer

2.       Message maintains their sequence even if consumer thread restarts in between

3.       Message Priority is IGNORED and sequenced is maintained 

4.       Results matches the documentation provided here: http://activemq.apache.org/exclusive-consumer.html

Test Case 2:

Messages are going through multiple hops before get consumed from exclusive consumer.

Result:

1.       Message loses sequence

2.       Priority Set on the messages resets to default (‘4’) regardless of producing value.

Summary:

·         If producer and consumer maintained in same broker and consumers are made exclusive; message sequencing is guaranteed.

·         If message is making multiple hops before reaching to consumer, sequencing is NOT guaranteed even if Exclusive consumers are used.

 In Part II, we will use JMXGroupId concept for message sequencing and find the right results.

Virtual Topic SelectorAware Consumers

Making a Consumer selectorAware comes with a lackuna.

If selectorAware property is r’true’, by default, consumer queue will not pull the message from the Topic, unless there is atleast one ‘Active Consumer Application’ listening to consumer queue. Even if there is no selector defined on the consumer queue.

On the contrary, if the selectorAware is false, all the messages from the Topic will be posted to consumer queue regardless of consuming application availability.

So, its actually a design decision to use this flag wisely for proper implementation.

Making Virtual Topic Consumer “Selector Aware”

Making Virtual Topic Consumer “Selector Aware”

<virtualDestinations>

<virtualTopic name="MF.PRICE.CHANGE.TP01" prefix="VTCON.*." selectorAware="true"/>

</virtualDestinations>

Making Virtual topics selector aware is as easy as flipping the flag to “true”.  For this example we created two consumers, one - “VTCON.GEN.MF.PRICE.CHANGE.TP01” listens to all the messages for ‘MF.PRICE.CHANGE.TP01’ topic and second – “VTCON.MKT.MF.PRICE.CHANGE.TP01" has a Selector of “Market = 100” on it.

Now, let’s send two messages, one with “market” string property with value 100 and another with value 200. In a result we should get one message in VTCON.MKT.MF.PRICE.CHANGE.TP01 as it has a Selector of “Market = 100” on it and two messages in “VTCON.GEN.MF.PRICE.CHANGE.TP01”

Java Code will look like below:

package test.consumer;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
public class VirtualTopicSelectorAwareConsumer {


static Connection connection;
static Session session;

public static void main(String[] args) {
ActiveMQConnectionFactory connectionFactory =
 new ActiveMQConnectionFactory("tcp://yourserver.com:port");

try {
 connection = connectionFactory.createConnection();
 session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
 // Consumer Listens to message with Market String property as 100
 String queueName = "VTCON.MKT.MF.PRICE.CHANGE.TP01";
 Destination mkt = session.createQueue(queueName);
 String selector100 = "Market = '100'";
 MessageConsumer consumer100 = session.createConsumer(mkt, selector100);
 Consumer100 listener = new VirtualTopicSelectorAwareConsumer().new Consumer100();
 consumer100.setMessageListener(listener);
 session.commit();
 // Consumer listens to all the message (Without a Selector)
 String genralConsumer = "VTCON.GEN.MF.PRICE.CHANGE.TP01";
 Destination genral = session.createQueue(genralConsumer);
 MessageConsumer genConsumer = session.createConsumer(genral);
 ConsumerNormal normal = new VirtualTopicSelectorAwareConsumer().new ConsumerNormal();
 genConsumer.setMessageListener(normal);
 session.commit();

 Topic topic = session.createTopic("MF.PRICE.CHANGE.TP01");
 MessageProducer producer = session.createProducer(topic);
 Message mktMessage = session.createTextMessage("Message for Market 100");
 mktMessage.setStringProperty("Market", "100");
 producer.send(mktMessage);

 Message mktMessage2 = session.createTextMessage("Message for Market 200");
 mktMessage.setStringProperty("Market", "200");
 producer.send(mktMessage2);
 producer.close();
 session.commit();

} catch (JMSException e) {
 e.printStackTrace();
}

}
/**
 * ConsumerNormal Listens All the messages and Log it
 */
private class ConsumerNormal implements MessageListener{

@Override
public void onMessage(Message arg0) {
 try {
  System.out.println(((TextMessage)arg0).getText());
  session.commit();
 } catch (JMSException e) {
  e.printStackTrace();
 }
}
}

/**
 * Consumer100 only listens to Market=100 messages
 */
private class Consumer100 implements MessageListener{
@Override
public void onMessage(Message arg0) {
try {
 System.out.println(((TextMessage)arg0).getText());
 session.commit();
} catch (JMSException e) {
 e.printStackTrace();
}
}
}
}

ActiveMQ Virtual Topics or Virtual Destinations

Creating Virtual Topics

Virtual Topics helps with following prospective:

1.       Load Balancing of messages

2.       Fast Failover of Subscriber

3.       Re-using same connection Factory for different Producers and Consumers. (Durable Subscribers needs a Unique JMS Client Id and same cannot be reused for any other Producer or consumer)

Steps to Define a Virtual Topic:

1.       Topic has to be created on Broker with Name: VirtualTopic.<TopicName>, However “VirtualTopic” is Out-of-the-box prefix for using this functionality.

2.       To make the change in prefix, following changes are needed.

<broker xmlns="http://activemq.apache.org/schema/core">

    <destinationInterceptors>

      <virtualDestinationInterceptor>

        <virtualDestinations>

          <virtualTopic name=">" prefix="VTCON.*." selectorAware="false"/>

        </virtualDestinations>

      </virtualDestinationInterceptor>

    </destinationInterceptors>

</broker>

In above settings two things are achieved.

A.      All Topics are made as Virtual Topic with name=">". However name field also accepts wildcard to apply different Virtual Topics policies. To change this Virtual Topic for Single Topic, just replace the “>” with actual Topic name.

<virtualDestinations>

<virtualTopic name="TEST.TP01" prefix="VTCON.*." selectorAware="false"/>

</virtualDestinations>

B.      Prefix for Consumer will be “VTCON”.  Now Virtual Topic consumers will be created with name like “VTCON.BASETBL.TEST.TP01”

C.      Settings are also made to NOT allow VirtualTopic Consumers as selectorAware. This will forward every message published to topic to consumers with “VTCON” prefix. If selectorAware is true, consumers will be able to add selector and only selected messages will be passed to consumers.