How to consume just one message from rabbit mq on nodejs

Question

Im using amqp.node library to integrate rabbitmq into my system.

But in consumer i want to process just one message at the time, then acknowledge the message then consume the next message from the queue.

The current code is:

// Consumer
open.then(function(conn) {
  var ok = conn.createChannel();
  ok = ok.then(function(ch) {
    ch.assertQueue(q);
    ch.consume(q, function(msg) {
      if (msg !== null) {
        othermodule.processMessage(msg, function(error, response){
          console.log(msg.content.toString());
          ch.ack(msg);
        });
      }
    });
  });
  return ok;
}).then(null, console.warn);

The ch.consume will process all messages in the channel at one time and the function of the module call it here othermodule will not be executed in the same time line.

I want to wait for the othermodule function to finish before consume the next message in the queue.

Answer 1

At this moment (2018), I think RabbitMQ team has an option to do that:

https://www.rabbitmq.com/tutorials/tutorial-two-javascript.html

ch.prefetch(1);

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.

Answer 2

Follow up the example here :

https://www.npmjs.com/package/amqplib

// Consumer
function consumer(conn) {
  var ok = conn.createChannel(on_open);
  function on_open(err, ch) {
    if (err != null) bail(err);
    ch.assertQueue(q);
    
     // IMPORTANT
    ch.prefetch(1);

    ch.consume(q, function(msg) {
      if (msg !== null) {
        console.log(msg.content.toString());
        ch.ack(msg);
      }
    });
  }
}

Refs: http://www.squaremobius.net/amqp.node/channel_api.html#channel_prefetch

Answer 3

You need to set a prefetch value as shown in this example:

https://github.com/squaremo/amqp.node/blob/master/examples/tutorials/rpc_server.js#L22