Configure how message distribution happens when prefetch are in use

[Lucas Rolff]

Hi everyone,

I have an application where we receive quite a few messages in RabbitMQ, which are prefetched in batches of 1000 messages per consumer - this works great overall when our queue is having a larger backlog, since we're always reaching the prefetch_count, and thus getting optimum performance out of the consumers.

However, if we're having e.g. 3 consumers listening on a queue, and there's 100 messages in the queue - each consumer receives 1/3 of the messages if they all happen to be "idle" when the 100 messages are published to the queue.

This means instead of a consumer receiving 100 messages (1/10 of the prefetch size), it only receives 33 or 34 messages, and this happens to all 3 workers.

Is there a way to configure RabbitMQ, to not distribute the messages evenly between the X number of workers, as long as the prefetch_count haven't yet been reached?

This would allow our application to become more effective, since we process all messages in one go, and return ack/nack for all messages in the batch at once (This is a lot more efficient in our case) - so obviously we want to keep the number of messages as high as possible as long as it stays below the prefetch_count.

Currently we start or stop consumers depending on the queue size to avoid this "problem", however it would be awesome if there's some way to configure the distribution logic when prefetch_count are greater than 1.

So if we take an example:

queue size: 1000 messages

prefetch_count: 1000

consumer_count: 3

In this case, if all 3 consumers are idle, two consumers will receive 333 messages and one consumer 334 messages.

If there's 4000 messages, all consumers obviously receive 1000 messages each and there's 1000 left in the queue.

What I would like to happen:

queue size: 1000 messages

prefetch_count: 1000

consumer_count: 3

What I would like to happen is that consumer0 gets 1000 messages (due to the prefetch_count being 1000) and consumer1 and consumer2 gets 0 messages.

I know generally speaking you want to distribute messages evenly between workers connected to your queue, so giving 333 messages to 2 workers and 334 messages to 1 worker makes sense in most scenarios, but allowing a consumer to receive as many messages (up to prefetch_count) can be useful in cases where one uses explicit acknowledgement of messages and do batch operations inside the consumers.

What we're doing (background on why we would like to control this 'distribution method'):

We have a system where some parts of the systems are highly distributed, and some receivers that are by no means distributed.

We have hundreds of thousands of websites sending json strings, each string contains a list of URLs (images, PDFs) and a "type" for each URL - we read the request, and depending on the type of a given URL, we publish a message to RabbitMQ for each and every URL.

This means if a website sends us a json containing 400 items, we queue 400 messages onto RabbitMQ - this is done because each item can take anywhere from 0.1 second to 300 seconds to process.

We then have N number of consumers listening for messages on this queue, these consumers have a prefetch_count of 1, and do whatever work is required to happen depending on some metadata in the message - when the consumer is done, it publishes a message back on a "notify" queue.

This notify queue has N workers, and will prefetch 1000 messages and based on a "tag" (an integer that refers to an item in an KV store on the notify consumers) in each message, we group the messages based on this integer and then do a POST request back to the website (establish connections, handle high latency, slow responses etc).

Since we're notifying websites, we want to batch up to a 1000 items back to a single website (in reality it's usually 200-250 items per site), this is simply done to lower the number of POST requests we have to do within the consumer, and thus increasing the throughput rate of messages (and not killing external services when we can bulk notify them).

Now, this is obviously great if our queue is 3000+ messages and we have 3 workers, since each worker will get 1000 items each.

However, when we only have 100 messages, this is distributed between 3 workers instead of 1, so even if all 100 messages belong to a single site, we end up doing 3 POST requests to the external service, when we could keep it at 1 request, if RabbitMQ allows to say: "Send all messages to X consumer if prefetch_count isn't reached".

1 vs 3 POST requests obviously isn't a big deal, but if we'd have 100 consumers instead of 3, we suddenly end up doing 100 calls instead of 1 call - so the "damage" scales with the number of consumers, due to the even distribution.

Sorry for the long post, but I also wanted to give some insight into why we ideally want to be able to control this distribution behavior when prefetch_count are higher than 1.

Best Regards,
Lucas Rolff

[jo...@cloudamqp.com]

Hi,

I think the closest to a solution would be to use a single active consumer, see [1] and [2].

[1] https://www.cloudamqp.com/blog/rabbitmq-3-8-feature-focus-single-active-consumer.html

[2] https://www.rabbitmq.com/consumers.html#single-active-consumer

[Lucas Rolff]

Then your throughput maxes out at a single consumer, which surely is fine as long as a single consumer can keep up. For that, I wouldn't need to use a 3.8 feature, after all I can just have a single consumer connected.

The issue lies in when you have multiple consumers (to handle higher workloads). The issue being the current distribution way isn't super optimal.

I guess I could use the consistent-hashing plugin and scale the number of queues as more processing capacity is needed.

Thanks!