Getting Start with RabbitMQ

Photo by Gary Bendig on Unsplash

What are Message Queues

Message queues are a form of asynchronous service-to-service communication used in server-less and micro-services architectures. First of all, messages are pushed onto the queue. One-by-one, they’re processed and consumed. Each message can only be processed once by an individual consumer. Message queues are very useful when decoupling concentrated processing, to batch work, and to facilitate heavy workload.

AMQP

The Advanced Message Queuing Protocol (AMQP) is an open standard for passing business messages between applications or organizations.

Let’s look at the key components of AMQP below.
Producer: Emits messages to the exchange
Consumer: Recieves messages from the queue
Binding: Connects an exchange with a queue using the binding key
Exchange: Compares routing keys with the binding key
Exchange Types: Determine the type of message distribution
Default Exchange: Compares routing key directly with queue name

Photo by Melanie Pongratz on Unsplash

What is RabbitMQ

RabbitMQ is an open-source message broker that implements AMQP. It allows for asynchronous communication and can be deployed in a distributed manner. It supports TLS and LDAP. It also comes with a big set of tools and plugins. It also has a GUI-based admin web-app where you can create channels and queues and even send and recieves messages. This helps the developer easily monitor and manage the workflow.
RabbitMQ has a lot of pros; let’s look at them below.

1. RabbitMQ allows dynamic routing. You can route a message to single queue, multiple queues or even selected queues. This ensures that messages are delivered to the correct device or service.
2. RabbitMQ can easily handle a large number of messages and connections making it a no-brainer for high-traffic apps.
3. RabbitMQ has a large and active community, which means that the internet can save you when you’re stuck.

Using RabbitMQ with Docker

Pull the Docker image and run it:

docker pull rabbitmq:3-management
docker run --rm -it -p 15672:15672 -p 5672:5672 rabbitmq:3-management

Assuming that it was installed correctly and is now running successfully, you’ve got a live instance of RabbitMQ! Head over to http://localhost:15672 to check out the admin web app.

Producer Code

Photo by Todd Cravens on Unsplash

We’re now heading towards the coding side of things so first of all, it’s very important for you install the AMQP library. It’s pretty simple. Just the run the following command:

npm install amqplib

Now let’s look at how the producer works. First of all, we will get the AMQP library. The next step is adding a sample message. In our code we’re taking direct command line input. After that we’ll write the connect() function. Enclose everything in the function inside a try-catch block. First, we will use amqp.connect() to connect the RabbitMQ server. After that, we have to create a channel. Now we will create a new queue called JobQueue. We will then send our message to that queue and close the channel and the connection. Now our JSON.stringified message is on the RabbitMQ server waiting to be fetched by our Consumer.

const amqp = require("amqplib");

const message = {number: process.argv[2]}

connect();

async function connect() {

    try {
        const AMQPServer = "amqp://localhost:5672"
        const connection = await amqp.connect(AMQPServer)
        const channel = await connection.createChannel();
        await channel.assertQueue("MessageQueue");
        await channel.sendToQueue("MessageQueue", Buffer.from(JSON.stringify(message)))
        console.log(`Job sent successfully ${message.number}`);
        await channel.close();
        await connection.close();
    }
    catch (ex){
        console.error(ex)
    }

}

Consumer Code

Most of the stuff in the consumer is pretty much the same until you get to the consumption part. Here, we assert the queue again and then consume all the messages in the queue. We can also set up acknowledgement of the message using channel.ack(). This will send a message back to the server, letting it know that a certain message has been consumed. Finally, we keep the consumer running continously so that it doesn’t miss any messages.

const amqp = require("amqplib");

connect();
async function connect() {

    try {
        const AMQPServer = "amqp://localhost:5672"
        const connection = await amqp.connect(AMQPServer)
        const channel = await connection.createChannel();
        await channel.assertQueue("JobQueue");
        
        channel.consume("JobQueue", message => {
            const input = JSON.parse(message.content.toString());
            console.log(`Recieved messsage with input ${input.number}`)
   
            //this is an example 
            if (input.number == 11) 
                channel.ack(message);
        })

        console.log("Waiting for messages...")
    
    }
    catch (ex){
        console.error(ex)
    }

}

Conclusion

Photo by Paula Hayes on Unsplash

Conclusion