CODEWITHSUNDEEP
node.jsperformancerabbitmqmessage-queue
Alex_Crack
Alex_Crack

Reputation: 710

RabbitMQ: how to limit consuming rate

I need to limit the rate of consuming messages from rabbitmq queue.

I have found many suggestions, but most of them offer to use prefetch option. But this option doesn't do what I need. Even if I set prefetch to 1 the rate is about 6000 messages/sec. This is too many for consumer.

I need to limit for example about 70 to 200 messages per second. This means consuming one message every 5-14ms. No simultaneous messages.

I'm using Node.JS with amqp.node library.

Upvotes: 19

Views: 34614

Answers (5)

Ahmed Adewale
Ahmed Adewale

Reputation: 3143

This is how I fixed mine with just settimeout

I set mine to process consume every 200mls which will consume 5 data in 1 seconds I did mine to do update if exist

channel.consume(transactionQueueName, async (data) => {
   let dataNew = JSON.parse(data.content);
       const processedTransaction = await seperateATransaction(dataNew);
        // delay ack to avoid duplicate entry !important dont remove the settimeout
        setTimeout(function(){
          channel.ack(data);
        },200);
 });

Done

Upvotes: 1

amcdnl
amcdnl

Reputation: 8648

See 'Fair Dispatch' in RabbitMQ Documentation.

For example in a situation with two workers, when all odd messages are heavy and even messages are light, one worker will be constantly busy and the other one will do hardly any work. Well, RabbitMQ doesn't know anything about that and will still dispatch messages evenly.

This happens because RabbitMQ just dispatches a message when the message enters the queue. It doesn't look at the number of unacknowledged messages for a consumer. It just blindly dispatches every n-th message to the n-th consumer.

In order to defeat that we can use the prefetch method with the value of 1. This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy.

Upvotes: 4

John Culviner
John Culviner

Reputation: 22974

Implementing a token bucket might help: https://en.wikipedia.org/wiki/Token_bucket

You can write a producer that produces to the "token bucket queue" at a fixed rate with a TTL on the message (maybe expires after a second?) or just set a maximum queue size equal to your rate per second. Consumers that receive a "normal queue" message must also receive a "token bucket queue" message in order to process the message effectively rate limiting the application.

NodeJS + amqplib Example:

var queueName = 'my_token_bucket';
rabbitChannel.assertQueue(queueName, {durable: true, messageTtl: 1000, maxLength: bucket.ratePerSecond});
writeToken();

function writeToken() {
    rabbitChannel.sendToQueue(queueName, new Buffer(new Date().toISOString()), {persistent: true});
    setTimeout(writeToken, 1000 / bucket.ratePerSecond);
}

Upvotes: 13

Alex_Crack
Alex_Crack

Reputation: 710

I've already found a solution.

I use module nanotimer from npm for calculation delays.

Then I calculate delay = 1 / [message_per_second] in nanoseconds.

Then I consume message with prefetch = 1

Then I calculate really delay as delay - [processing_message_time]

Then I make timeout = really delay before sending ack for the message

It works perfectly. Thanks to all

Upvotes: 10

Vor
Vor

Reputation: 35149

I don't think RabbitMQ can provide you this feature out of the box.

If you have only one consumer, then the whole thing is pretty easy, you just let it sleep between consuming messages.

If you have multiple consumers I would recommend you to use some "shared memory" to keep the rate. For example, you might have 10 consumers consuming messages. To keep 70-200 messages rate across all of them, you will make a call to Redis, to see if you are eligible to process message. If yes, then update Redis, to show other consumers that currently one message is in process.

If you have no control over consumer, then implement option 1 or 2 and publish message back to Rabbit. This way the original consumer will consume messages with the desired pace.

Upvotes: 1

Related Questions