Distributed Message Queues

Message queues are an important category of indirect communication systems.

In indirect communication, both time and space are uncoupled. That means, sender and receiver don’t need to exist at the same time to communicate and sender doesn’t know receivers identity vice versa.

Since message queues belong to indirect communication, message queues provide a point-to-point service achieving the particular properties. They are point-to-point in that the sender places the message into a queue, and it is then removed by a single process.

Message queues avoid doing a resource-intensive task immediately and wait for it to complete. Instead we schedule the task to be done later.

Producer: A program that sends a message.

Consumer: A program that mostly waits to receive a message.

Create a message queue in javascript with RabbitMQ

For this, we need to install follwings,

RabbitMQ

Installing on Windows

ERlang

Erlang Programming Language

We need RabbitMQ server running on localhost on standard port 5672.

Get Mahesha Muthumala’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Then, we need to install amqp.node client library using npm:

npm install amqplib

Now, we will need two javascript files as sender and receiver.

sender.js

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
  if (error0) {
    throw error0;
  }
  connection.createChannel(function(error1, channel) {
    if (error1) {
      throw error1;
    }
    var queue = 'task_queue';
    var msg = process.argv.slice(2).join(' ') || "Hello World!";

    channel.assertQueue(queue, {
      durable: true
    });
    channel.sendToQueue(queue, Buffer.from(msg), {
      persistent: true
    });
    console.log(" [x] Sent '%s'", msg);
  });
  setTimeout(function() {
    connection.close();
    process.exit(0)
  }, 500);
});

receiver.js

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(error0, connection) {
  if (error0) {
    throw error0;
  }
  connection.createChannel(function(error1, channel) {
    if (error1) {
      throw error1;
    }
    var queue = 'task_queue';

    channel.assertQueue(queue, {
      durable: true
    });
    channel.prefetch(1);
    console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);
    channel.consume(queue, function(msg) {
      var secs = msg.content.toString().split('.').length - 1;

      console.log(" [x] Received %s", msg.content.toString());
      setTimeout(function() {
        console.log(" [x] Done");
        channel.ack(msg);
      }, secs * 1000);
    }, {
      // manual acknowledgment mode,
      // see https://www.rabbitmq.com/confirms.html for details
      noAck: false
    });
  });
});

We need to run these files in different shells and we need to run the receiver.js file again in a new shell to start a new consumer.

./sender.js
./receiver.js

Message acknowledgement

RabbitMQ supports message acknowledgments to make sure a message is never lost. An ack(nowledgement) is sent back by the consumer to tell RabbitMQ that a particular message has been received, processed and that RabbitMQ is free to delete it.

If a consumer dies without sending an ack, RabbitMQ will understand that a message wasn’t processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer.

That way we can be sure that no messages is lost, even if the consumers occasionally die.

Message Durability

We declared the queue and messages as durable, in order to make sure that the queue will survive when RabbitMQ quits or crashes.