RabbitMQ have an exclusive consumer consume message serially

Question

I have a scenario that on a given topic I need to consume each message one by one, do some async task and then consume the next one. I am using rabbitmq and amqp.node.

I was able to achieve this with a prefetch of 1. Which of course is not an actual solution since this would lock the whole channel and the channel have multiple topics.

So far this is my producer:

const getChannel = require("./getChannel");
async function run() {
    const exchangeName = "taskPOC";
    const url = "amqp://queue";
    const channel = await getChannel({ url, exchangeName });
    const topic = "task.init";
    let { queue } = await channel.assertQueue(topic, {
        durable: true
    });
    const max = 10;
    let current = 0;
    const intervalId = setInterval(() => {
        current++;
        if (current === max) {
        clearInterval(intervalId);
        return;
        }
        const payload = JSON.stringify({
        foo: "bar",
        current
        });

        channel.sendToQueue(queue, Buffer.from(payload), { persistent: true });
    }, 3000);

}
run()
.then(() => {
    console.log("Running");
})
.catch(err => {
    console.log("error ", err);
});

And this is my consumer

const getChannel = require("./getChannel");
async function run() {
    const exchangeName = "taskPOC";
    const url = "amqp://queue";
    const channel = await getChannel({ url, exchangeName });
    channel.prefetch(1);
    const topic = "task.init";
    const { queue } = await channel.assertQueue(topic, {
        durable: true
    });
    channel.bindQueue(queue, exchangeName, topic);
    let last = new Date().getTime();
    channel.consume(
        queue,
        msg => {
        const now = new Date().getTime();
        console.log(
            " [x] %s %s:'%s' ",
            msg.fields.routingKey,
            Math.floor((now - last) / 1000),
            msg.content.toString()
        );
        last = now;
        setTimeout(function() {
            channel.ack(msg);
        }, 10000);
        },
        { exclusive: true, noAck: false }
    );
}
run()
.then(() => {
    console.log("Running");
})
.catch(err => {
    console.log("error ", err);
});

Is there any way on RabbitMQ to do that or I would need to handle this on my app?

Thanks.

cdelmas cdelmas · Accepted Answer · 2018-12-18T12:43:51

You can use the consumer prefetch setting (see https://www.rabbitmq.com/consumer-prefetch.html). In the case of amqp node, you set this option using the prefetch function:

channel.prefetch(1, false); // global=false

In this case, each consumer on the channel will have a prefetch of 1. If you want to have different configurations for each consumer, you should create more channels.

Hope this helps.

RabbitMQ have an exclusive consumer consume message serially

1 Answers