Performance Zone is brought to you in partnership with:

I'm a Sr Software Engineer working at RedHat on open source messaging solutions. I've been working now for over ten years on a variety of software projects. I started working on open source full time in 2007 and have contributed a great deal of work to the ActiveMQ project. Timothy is a DZone MVB and is not an employee of DZone and has posted 7 posts at DZone. You can read more from them at their website. View Full User Profile

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

07.25.2013
| 4809 views |
  • submit to reddit
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.

Table 2. Settings for Abort Slow Consumer Strategy
Attribute Default Description
maxTimeSinceLastAck 30000 Specifies the amount of time that must elapse since the last message acknowledge before a consumer is marked as slow.
ignoreIdleConsumers true Specifies whether the time since last acknowledge is applied to consumer that has no dispatched messages.  If set false this strategy can be used to abort any consumer after the time since last acknowledge interval has expired.
maxSlowCount -1 Specifies the number of times a consumer can be considered slow before it is aborted. -1 specifies that a consumer can be considered slow an infinite number of times.
maxSlowDuration 30000 Specifies the maximum amount of time, in milliseconds, that a consumer can be continuously slow before it is aborted.
checkPeriod 30000 Specifies, in milliseconds, the time between checks for slow consumers.
abortConnection false Specifies whether the broker forces the consumer connection to close. The default value specifies that the broker will send a message to the consumer requesting it to close its connection. true specifies that the broker will automatically close the consumer's connection.

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>


Published at DZone with permission of Timothy Bish, author and DZone MVB. (source)

(Note: Opinions expressed in this article and its replies are the opinions of their respective authors and not those of DZone, Inc.)