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>

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.

[INFO] Cannot find lifecycle mapping for packaging: 'bundle'.
Component descriptor cannot be found in the component repository: org.apache.maven.lifecycle.mapping.LifecycleMappingbundle.

The fix it, set the ‘extensions’ element to true as shown below.

<plugin>
                <groupId>org.apache.felix</groupId>
                <artifactId>maven-bundle-plugin</artifactId>
                <version>2.0.1</version>
                <extensions>true</extensions>
                <configuration>
                    .... 
 
                </configuration>
</plugin>

Hope this saves you some digging!

11/19/08 5:22:23 PM EST: resources:resources
11/19/08 5:22:23 PM EST: ERROR reactor-execute : /home/davestanley/Work/eclipse/cxf-se/TestTwo : org/codehaus/plexus/util/FileUtils$FilterWrapper

The workaround is to add the maven-resources-plugin to your projects pom.xml.

1
2
3
4

 <plugin>                
           <artifactId>maven-resources-plugin</artifactId>
           <version>2.2</version>
 </plugin>

In order to add the camel-hibernate plugin as a dependency of your camel project you just need to add the repo:

1
2
3
4
5

<repository>  
  <id>camel-extra-release</id>  
  <name>Camel Extra Maven repository of releases</name>  
  <url>http://camel-extra.googlecode.com/svn/maven2/releases</url>
</repository>

Then add the dependency for the hibernate component:

1
2
3
4
5

<dependency>  
   <groupId>org.apache.camel.extra</groupId>  
   <artifactId>camel-hibernate</artifactId>  
   <version>1.0</version>
</dependency>

(Thanks Joe for the heads up)

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.