RabbitMQ suspend queue consumption

Question

What is the best way to keep a durable queue and its bindings but suspend its consumers?

The use case is: I would like to "let it crash" and stop handling messages if we keep getting a bunch of messages we can't deal with (e.g. the database is down or schema issue) but would like to keep aggregating to the queue. That is allow publish but suspend consuming.

I can think of three solutions:

1. I could have all the consumers bound to the queue continuously reject the messages and re-queue but this is sort of waste resources not to mention I have programmatically do the above logic.
2. I could call basic.cancelConsumer on all the consumers (see below)
3. Or in terms of spring-amqp I guess I could call shutdown on all the SimpleMessageListenerContainers that are bound to the queue.

#1 we are already doing as the messages are getting rejected. The problem is this ends up being like an infinite while loop of failure and if your logging that failure even more resources are wasted.

#3 seems ideal but I have to some how have know about all the message listeners and then notify them to shutdown. I suppose I could use a fanout exchange to notify that the queue needs to be suspended. I feel like RabbitMQ must have something built-in for this logic. The other issue is that you can bind multiple queues to a message container (not all queues may need to be suspended).

For #2 I know I can cancel the consumer with its consumerTag but the question (assuming that is the right way to do the above) is where do I get the list of consumerTags to a queue?

Solution 1

If you need to stop a consumer then just call basic cancel, that's the way to do it.

Having durable queues is a matter that you solve when you declare the queue: durable=true auto_delete=false.

Having persistent messages is determined when you publish them: delivery_mode=2.

Each consumer has it's own consumer tag. Whenever you get a message out of the queue, the envelope should have the consumer tag and you should call basic cancel with that tag.

AFAIK, you can't have consumer A in conn 1 and call basic cancel for that consumer on a different connection. I might be wrong on this one.

Solution 2

This solution is spring-amqp specific. Its basically #3 from my answer.

I maintain a service that has a Map<String,CustomSimpleMessageListenerContainer> of queue name to custom extended SimpleMessageListenerContainer.

After a certain amount of exceptions I send a "panic" message which goes to a special queue received by the service to shutdown the consumer.

Comments

[Adam Gent]

Thanks. I know how to declare queues in rabbit like above. It looks like I'll have to manage my consumers on my own which is the path I'm already going down. It would be nice if rabbit had the option admin wise to suspend consumption of a queue.

[old_sound]

The only thing you can do from a connection is to force close it from the admin which is not really what you are looking for

[watery]

The panic delivered with a secondary queue is a nice idea!