Coming in ActiveMQ 5.9 a new way to abort slow consumers.

The topic of how to deal with a slow consumer in ActiveMQ comes up on the mail list from time to time.  In the current releases there is a strategy class that allows you to configure a time interval after which a slow consumer is closed (some nice documentation can be found here).  The AbortSlowConsumerStrategy detects a consumer is slow by looking at its prefetch buffer and checking if its been to full for to long. When the consumer is aborted the messages that were in its prefetch buffer can then be sent on to other consumers on the same destination.

The AbortSlowConsumerStrategy works well when you have consumers whose prefetch limit is set to a value greater than one, but for those cases where you've set a value of zero or one the consumer may never get marked as slow by the broker so the strategy might never kick in to abort the slow consumers. 

To solve the problem of slow consumers with small prefetch values we've introduced the AbortSlowAckConsumerStrategy.  This strategy works by checking the time since a consumer last acknowledged a message as apposed to the time that the consumer has had a full prefetch buffer.  The strategy will poll all the consumers of destination at a configurable rate to determine if the consumer is slow and if the consumer meets the configured criteria for abort then it will be given the boot and any prefetched messages will be redispatched. 

The AbortSlowAckConsumerStrategy is configured in the XML configuration file as follows:

<broker ... >
  ...
  <destinationPolicy>
    <policyMap>
      <policyEntries>
        <policyEntry topic=">" >
          <slowConsumerStrategy>
            <abortSlowConsumerStrategy />
          </slowConsumerStrategy>
        </policyEntry>
      </policyEntries>
    </policyMap>
  </destinationPolicy>
  ...
</broker>

This configuration adds the AbortSlowAckConsumerStrategy with default options and applies it only to Topic consumers, for Queues you just need to replace the word 'topic' with 'queue' in the policy entry.

There are various options that can be set on the strategy to determine when it aborts a slow consumer.


If you read through the options above you should notice that it's possible to use this strategy not only to abort slow consumers that are receiving messages from the broker but also you can use this as a way to abort any consumer after a specified time interval. To accomplish this all you need to do is set the ignoreIdleConsumers option to false and the strategy will treat any consumer that hasn't acknowledged a message since the configured maxTimeSinceLastAck regardless of whether its had a message dispatched to it or not.

Lets look at an example of how to abort any topic consumer that hasn't acknowledged a message in the last 60 seconds:

<broker ... >
  ...
  <destinationPolicy>
    <policyMap>
      <policyEntries>
        <policyEntry topic=">" >
          <slowConsumerStrategy>
            <abortSlowConsumerStrategy ignoreIdleConsumers="false" maxTimeSinceLastAck="60000" />
          </slowConsumerStrategy>
        </policyEntry>
      </policyEntries>
    </policyMap>
  </destinationPolicy>
  ...
</broker>