Dispatch PoliciesActiveMQ 4.0 or later supports the configuration of different dispatch policies on a per destination (or wildcard) basis. Before discussing dispatch policies its worth first understanding the purpose of the prefetch value. The out of the box configuration of ActiveMQ is designed for high performance and high throughput messaging where there are lots of messages that need to be dispatched to consumers as quickly as possible. So the default prefetch values are fairly large and the default dispatch policy will try and fill the prefetch buffers as quickly as possible. However with messaging there are many use cases and sometimes the default configuration is not ideal to your use case; when you send a small number of messages, they tend to all go to one consumer unless you've lots of messages. If you have a large number of consumers and a relatively high prefetch value and you have a small number of messages that each message takes quite a while to process then the default dispatch policy might result in increasing the amount of time it takes to process all the messages (since the load balancing is not fair for small numbers of messages). Round Robin PolicyTo load balance low numbers of messages more fairly you might want to switch to the <roundRobinDispatchPolicy/> which tries to dispatch messages fairly at the expense of some performance. Here is an example of this in use. <beans> <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/> <broker persistent="false" brokerName="${brokername}" xmlns="http://activemq.apache.org/schema/core"> <!-- lets define the dispatch policy --> <destinationPolicy> <policyMap> <policyEntries> <policyEntry topic="FOO.>"> <dispatchPolicy> <roundRobinDispatchPolicy /> </dispatchPolicy> <subscriptionRecoveryPolicy> <lastImageSubscriptionRecoveryPolicy /> </subscriptionRecoveryPolicy> </policyEntry> <policyEntry topic="ORDERS.>"> <dispatchPolicy> <strictOrderDispatchPolicy /> </dispatchPolicy> <!-- 1 minutes worth --> <subscriptionRecoveryPolicy> <timedSubscriptionRecoveryPolicy recoverDuration="60000" /> </subscriptionRecoveryPolicy> </policyEntry> <policyEntry topic="PRICES.>"> <!-- 10 seconds worth --> <subscriptionRecoveryPolicy> <timedSubscriptionRecoveryPolicy recoverDuration="10000" /> </subscriptionRecoveryPolicy> <!-- lets force old messages to be discarded for slow consumers --> <pendingMessageLimitStrategy> <constantPendingMessageLimitStrategy limit="10"/> </pendingMessageLimitStrategy> </policyEntry> </policyEntries> </policyMap> </destinationPolicy> </broker> </beans> |