Aim

To use a messaging middleware queue to distribute the workload across the jvm is quiet common and helps in linear scalability. This blog explain the design pattern “Polling Consumer” which aids in workload distribution. This blog also explains how the messaging middleware itself become a bottleneck in linear scalability and how to overcome it. Quiet a famous saying I have heard in many of design meetings "Drop a message in a queue, it will be picked up by the consumer to perform the processing. Increase the consumers to increase the performance". Lets see how much truth in it :-)

Polling Consumer

In any messaging system (queue), the consumer needs an indication that application is ready so that it can consume the message. The best approach for the consumer is to repeatedly check the channel for message availability. If any message is available, the consumer consumes it (process it). This checking is a continuous process known as polling. The messaging system (queue) can be polled from the multiple jvms. All the consumers/jvms should equally share the workload and should not keep hold of messages on which they are not working.

Solution:

The application should use a polling consumer, one that explicitly makes a call when it wants to receive a message.

Interactions:

A polling consumer is a message receiver. A polling consumer restricts the number of concurrent messages to be consumed by limiting the number of polling threads. In this way, it prevents the application from being blocked by having to process too many requests, and keeps any extra messages queued up until the receiver can process them.Each consumer/jvm will ask for message whenever it has capacity to perform hence provide perfect load balancing automatically.
Merits

Consumer asks for work (as a message) whenever it has capacity to perform, hence provide efficient workload management.
No message is lost as all the messages are persisted in the messaging system (queue).
Increasing number of consumers (jvms) can increase the scalability/performance to certain level.

Demerits
Predicable scalability/performance cannot be guaranteed, as increasing consumers (jvms) the messaging system can become the bottleneck as shown by the picture below

The messaging middleware (queue) becomes the bottleneck because, as the numbers of consumers are increased to achieve the linear scalability, the messaging middleware fails to handle so many consumers. The messaging middleware will look like a overloaded horse cart like below

Poor horse chart or should I say messaging middle ware ;-)

The following picture shows the decreasing performance (i.e. no messages processed per seconds) as the numbers of jvms are increased. In below the queue used is a durable queue.

NOTE: - The following tests were performed on windows machine.

Hence with queue in db after the performance reached the optimum level (5 jvms as shown in above picture), adding more jvm results in negative performance.

Predictable Scalability

One the key feature of Oracle Coherence is to provide “Predictable Scalability” i.e. by increasing a node (a jvm) in a cluster the performance of the system increases. The following picture shows the Queue of the “Polling Consumer” pattern being implemented using Oracle Coherences + AQ that is, AQ being made as the persistent store for the oracle coherence. As a result the consumer always enqueue/dequeue messages from the in-memory queue (hence AQ –DB bottleneck is removed). The enqueue will result a message in the coherence cache and also AQ and similarly dequeue of the message will result in message being removed from the coherence cache and AQ.

In the above design the cache store make sure message are always in sync with persistent storage (db).

1. enqueue => oracle coherence cache put(message) => AQ enqueue

2. dequeue=> oracle coherence cache remove() => AQ dequeue

So the endpoint always perform operation in the cache

Conclusion

A queue is an efficient way distributing work load. But the linear scalability cannot be always assured.ln functional areas like invoice processing,claim processing etc, using queue for workload distribution may not be always helpful in linear scalability. If the number of the JVMs are less then desired performance can be achieved by fine tunning the consumers.

In my next blog I will share the framework i.e design details to achieve the above. The "Polling Consumer" is an enterprise integration pattern (http://www.eaipatterns.com/PollingConsumer.html). In past, as a part of my job I have developed spring based framework which enable the implementation of enterprise integration design pattern. But spring source team have added a new framework called "Spring Integration", which enable ease of implementing enterprise integration pattern. How to create "Polling Consumer" using spring integration can be found at http://static.springsource.org/spring-integration/reference/htmlsingle/spring-integration-reference.html#endpoint-pollingconsumer

A sneak preview - in order to achieve Polling Consumer backed by queue in a cache (coherence) which also persist messages we need the following

1) Off course, a queue/channel in oracle coherence backed by persistence storage. Lets call it - CoherenceAwareQueueChannel

2) A Oracle Coherence cache store which persist the messages

3) A simple POJO to which poller delivers the message - consumer. In the terminology of enterprise integration its called Service Activator (refer to http://www.eaipatterns.com/MessagingAdapter.html for detail)

Enqueue

Message message = MessageBuilder.withPayload("gaurav in cache").setHeader(JmsHeaders.TYPE, "sometype").build();

coherenceQueue.send(message);

Dequeue

Message msg = (Message) coherenceQueue.receive();

Polling Consumer


<beans:bean id="coherenceQueue" class="com.oracle.aq.cachestore.channel.CoherenceAwareQueueChannel" depends-on="cacheFactoryCreation"/>
<beans:bean id="messageConsumer" class="com.oracle.aq.cachestore.junit.AqCacheStoreTest$ServiceActivator"/>
<service-activator id="coherenceConsumer" input-channel="coherenceQueue"
 ref="messageConsumer" method="onMessage" auto-startup="false">
 <poller>
  <interval-trigger interval="10" />
 </poller>
</service-activator>

As a result of above a message will be passed to the POJO - ServiceActivator method onMessage. The poller will poll the CoherenceAwareQueue every 10 nano seconds. It's also possible to specify the number of polling threads.

The service activator pojo looks like
 public class ServiceActivator {
  
  public void onMessage(Message msg) {
                // Message will be delivered automatically
  }
 }

Simple programming model agnostic of coherence/spring details/jms details so that developer can use it with ease