Dave Stanley's Blog

Recently I’ve been working with ActiveMQ virtual destinations. In some cases it can be a really useful technique to scale up your topic consumers. I’ve posted my abbreviated notes below.

Virtual Destinations are logical destinations (i.e. Queues or Topics) that map onto one or more physical destinations. Their purpose is to overcome some load balancing/HA limitations with durable subscribers.

How it works:

1) Producer sends to the logical topic “MyTopic”

2) Consumer A can consume from the “MyTopic” as a normal Topic consumer.

3) Consumers B & C can also consume, via a dedicated physical queue linked to the logical topic. The link part is really what changes a regular topic into a virtual topic. So now we have a setup like this:

[producer] --> [MyTopic] --> [TopicConsumerA]
 
                        --> [VirtualQueueConsumer.MyTopic] --> [QueueConsumerB][QueueConsumerC]

Note: “VirtualQueueConsumer.MyTopic” is a physical queue. This means you can have a pool of consumers consuming from the queue, allowing you to overcome some of the limitations of durable subscribers.

Configuration for virtual destinations is via the brokers /conf/activemq.xml:

 
<destinationInterceptors> 
   <virtualDestinationInterceptor> 
     <virtualDestinations> 
 
   <!-- deliver traffic from virtual 'MyTopic' to all subscribers to destinations matching the prefix "VirtualQueueConsumer.*" (queue or topic) -->
   <virtualTopic name="MyTopic" 
                    prefix="VirtualQueueConsumer.*." /> 
     </virtualDestinations> 
    </virtualDestinationInterceptor> 
   </destinationInterceptors> 
 </broker>

In general, the out of the box ActiveMQ configuration works very well for the majority of usecases. One place where you will want to tweak things is in the area of supporting large numbers of concurrent connections.

The good news is that ActiveMQ is well capable and with the tweaks below you should be well on your way to supporting thousands of concurrent connections. Without further adieu, here’s what you need to do:

1) In your brokers /activemq.xml, enable the nio transport.

 
<!-- The transport connectors ActiveMQ will listen to -->
 <transportConnectors>
 
   <!-- use tcp port for network connectors only -->
   <transportConnector name="openwire" uri="tcp://localhost:61616"/>
 
   <!-- use nio port for producers/consumers -->
   <transportConnector name="openwire nio" uri="nio://localhost:62828?useQueueForAccept=false"/>
 
 </transportConnectors>

2) Again, in your brokers /activemq.xml, configure a destinationPolicy with optimizedDispatch=true, for example:

   <amq:destinationPolicy>
   <amq:policyMap>
    <amq:policyEntries>
     <amq:policyEntry queue=">"
       optimizedDispatch="true"
       lazyDispatch="false"
       producerFlowControl="false"
       memoryLimit="128 mb"
       strictOrderDispatch="true">
      <amq:dispatchPolicy>
       <amq:strictOrderDispatchPolicy />
      </amq:dispatchPolicy>
      <amq:messageGroupMapFactory>
       <amq:simpleMessageGroupMapFactory/>
      </amq:messageGroupMapFactory>
      <amq:subscriptionRecoveryPolicy>
       <amq:timedSubscriptionRecoveryPolicy recoverDuration="360000" />
      </amq:subscriptionRecoveryPolicy>
     </amq:policyEntry>
    </amq:policyEntries>
   </amq:policyMap>
  </amq:destinationPolicy>

For more info on optimizedDispatch see Hiram‘s article here

3) At the OS level increase the number of file descriptors available to the broker

>ulimit -n 4096

4) Use a broker with a version > 5.3.0.3

5) Tweak your TCP/IP layer to support the connection load

On some operating systems you may need to tune the tcp/ip stack to be able to handle the incoming connection load. Below are the settings I’ve found worked well for Linux and Solaris.

Linux tcp tuning settings:

sudo /sbin/sysctl -w net.core.netdev_max_backlog=3000
sudo /sbin/sysctl -w net.ipv4.tcp_fin_timeout=15
sudo /sbin/sysctl -w net.core.somaxconn=3000

Note: The net.core.netdev_max_backlog controls the size of the incoming packet queue for upper-layer (java) processing.

Solaris tcp tuning settings:

# ndd -set /dev/tcp tcp_fin_wait_2_flush_interval 67500
# ndd -set /dev/tcp tcp_keepalive_interval 30000
# ndd -set /dev/tcp tcp_conn_req_max_q 8000

Lastly, when testing I’ve found the ActiveMQ jmx mbeans and jconsole to be indispensable in terms of monitoring thread counts and the number of concurrent connections. If all is configured properly each 16 incoming connections should result in one new thread in the broker VM.

These minor changes should allow you to support several thousand concurrent connections on conventional hardware. If you happen to try the settings above let me know how it goes.

How fast you can send messages through ServiceMix is highly dependent on how the embedded ActiveMQ broker within ServiceMix is configured. This is because the ServiceMix NMR uses ActiveMQ as its messaging engine, so choosing the correct connection factories and persistenceAdapter can give you easy gains in performance.

AMQ Message Store

In ActiveMQ 5.0, a new high performance journaling persistence adapter was introduced called the AMQ Message Store.

To enable it, edit /conf/activemq.xml and modify the xbean config, adding the amq:persistenceAdapter element.

1
2
3
4
5
6
7
8

 <amq:persistenceAdapter>
         <!-- Goodbye journaledJDBC --> 
         <!--amq:journaledJDBC journalLogFiles="5" dataDirectory="./data/amq"/ -->

         <!-- Hello AMQ Message Store -->
         <amq:amqPersistenceAdapter directory="file://./data/amq"/>

 </amq:persistenceAdapter>

And by the numbers ..


Example Flow: Jms consumer -> JMS provider(w/Marshaller) -> EIP pipeline -> Bean -> JMS Provider
MessageCount: 10000

   With JournaledJDBC:

   [java] Overall time: 137701 ms
   [java] Messages per sec: 7.262111386264443
   [java]
   [java] Time: 139.603
   [java]
   [java] OK (1 test)

   With AMQ Message Store:

   [java] Overall time: 20511 ms
   [java] Messages per sec: 48.75432694651651
   [java]
   [java] Time: 23.556
   [java]
   [java] OK (1 test)

Two additional steps:

1) The current version of Servicemix trunk is using 4.1.1 of ActiveMQ, so if you want to build a ServiceMix distribution that contains ActiveMQ 5, you need to edit /trunk/pom.xml:

1
2
3

/trunk/pom.xml
-        <activemq-version>4.1.1</activemq-version>
+        <activemq-version>5.1.0</activemq-version>

2) Once the activemq-version is updated, the ra namespace in jndi.xml needs to be updated too – as this changed between AMQ versions.

/distributions/apache-servicemix/src/main/release/conf/jndi.xml
-       xmlns:amqra="http://activemq.org/ra/1.0"
+       xmlns:amqra="http://activemq.apache.org/schema/ra"

If your using Fuse ESB, it has a 5.x version of the Fuse Message Broker included, so the XML configuration change should be all thats required.

The Fuse September Release is now available. New in this release:



- Message Broker 5.0 (based on Apache ActiveMQ trunk)

- FUSE ESB 3.3 (Based on Apache ServiceMix 3.2 )

- Mediation Router 1.2 (based on Apache Camel 1.2)

- Services Framework 2.0.2 (based on Apache Incubator CXF 2.0.2)

(If your in an RSS Reader and swf doesn’t embed click here)

Update: Hmm, looks like DylanMessaging are having trouble. If you don’t see the message .. try again later!