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.

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>

Related

Trending

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

Related

Partner Resources