If not addressed, this can result in a disarray of communication among services, defeating the measures placed to ensure orderly database transactions. But thank goodness, we have message brokers to the rescue.

In this article, we'll highlight the importance of message queuing as a best practice for backend development, relevant use cases and popular message queuing tools, and how to implement message queuing in backend applications.

Here are some of the prerequisites to be able to follow along in this article:

• Knowledge of Node.js

• Knowledge of microservices architecture

What is Message Queuing?

In distributed systems, several requests and queues are sent at a time. The concept of a message queue enables the storage of messages in an orderly manner, allowing for the recipients of these messages and requests to process them accordingly.

It operates asynchronously, allowing the independent functioning of different components of a distributed system. Having these in place ensures that the messages sent to the recipient eventually get attended to irrespective of a system downtime. The messages are securely stored until they get acknowledged.

Relevant Use Cases of Message Brokers

Here are some of the real-life use cases of message brokers.

• They are actively used in modern fintech applications to ensure seamless and orderly execution and processing of financial transactions made on the application. This helps to prevent server overload and transaction errors.

• Message queueing is also used in our day-to-day notification applications, ensuring early reception of sent notifications from other services. This allows the recipient to get access to those notifications notwithstanding the time they were sent or when the recipient gets access to the notification application.

• It is also used in the financial markets for seamless and efficient execution of financial orders being made. Other uses of this feature are seen in media streaming and the healthcare industry.

In the next paragraph, we'll discuss more about the tools that offer message queueing features.

Examples of Popular Message Queue Services

A wide range of applications and services offer message queueing features. Some of these services are embedded in commercial cloud infrastructure providers. Here is a list of some commonly used message queueing services:

• RabbitMQ

• Apache Kafka

• Redis

• Amazon SQS

• Google Cloud Pub/sub

• NATS

• Pulsar

• IBM MQ

We'll be utilizing a Rabbit MQ Cloud-as-a-service application to power our messages due to its popularity and ease of use. Here is a link to the documentation. You can also check out other message queuing applications provided above.

Next, we'll develop a demo project that utilizes message queuing features.

Demo Project

In this project, we'll use Rabbit MQ as a service cloud platform to build a simple message broker system that allows for seamless, ordered communication between two Node.js servers.

In this tutorial, we'll create a message publisher that will serve as the sender, and a message consumer that receives the messages.

To begin with, we'll have to create both Node.js servers that will be communicating with each other.

You can create two different files and initialize a Node project using npm init.

Thereafter, you can install relevant packages that will aid in the implementation of the features. We'll use the amqplib library, a Node library implementation for Rabbit MQ.

This package allows us to swiftly communicate with RabbitMQ via the Node.js application. It seamlessly achieves this due to its built-in functions for creating queues, publishing messages, and consuming messages. More details regarding its usage will be discussed later.

To install this in our project, kindly execute:

npm i amqplib

The publish function will now be drafted. After that, we'll have to initialize amqplib in our project.

const amp = require(“amqplib”)

Also, we need to set up our RabbitMQ broker which will manage our messages.

There are several ways of creating RabbitMQ servers, the most popular being installing them on a home computer and then setting them up to interact with the backend servers. You can download the software here. However, for ease of usage, we will be utilizing a cloud-based RabbitMQ broker as a service application to generate our server.

To get this done, kindly navigate to https://www.cloudamqp.com/ and create an account. For this tutorial, an instance was created and configured to the closest region to me. On successful creation of the instance, the details of the Rabbit MQ will be made available.

Moving on, we will be creating a message queue in which both parties can use as a connection pipeline. We will begin by creating a function to send messages.

async function sendMessage(msg) {
  try {
    const connection = await amqp.connect(url);
    const channel = await connection.createChannel();

    await channel.assertQueue(queue);
    await channel.sendToQueue(queue, Buffer.from(msg));

In the code above, a connection was ensured and maintained. Thereafter, a communication channel was also created. The assert queue function is then declared when executed, ensuring that the existing queue is maintained, and creates the queue if it doesn't exist.

The message attached to the function gets buffered and then sent to the queue created.

async function receiveMessage() {
  try {
    const connection = await amqp.connect(url);
    const channel = await connection.createChannel();

    await channel.assertQueue(queue);
    await channel.consume(queue, (msg) => {
      console.log(`Received message: ${msg.content.toString()}`);
      channel.ack(msg);
    });

The receiver function also gets executed to receive any message that gets into the queue by executing the consume method at the exact queue. In our case, the message is outputted as a log message.

The ack function is now executed to acknowledge the message received from the queue.

Here is the full project code:

const amqp = require("amqplib");
const url = "amqp://localhost"; // Replace with your RabbitMQ server URL
const queue = "queue";

async function receiveMessage() {
  try {
    const connection = await amqp.connect(url);
    const channel = await connection.createChannel();

    await channel.assertQueue(queue);
    await channel.consume(queue, (msg) => {
      if (msg !== null) {
        console.log(`Received message: ${msg.content.toString()}`);
        channel.ack(msg);
      }
    });
  } catch (err) {
    console.error("Failed to receive message:", err);
  }
}

receiveMessage();

Here's the message publisher application code:

const amqp = require("amqplib");
const url = "amqp://localhost"; // Replace with your RabbitMQ server URL
const queue = "queue";

async function sendMessage(msg) {
  try {
    const connection = await amqp.connect(url);
    const channel = await connection.createChannel();

    await channel.assertQueue(queue);
    await channel.sendToQueue(queue, Buffer.from(msg));

    console.log(`Message sent to ${queue}: ${msg}`);

    await channel.close();
    await connection.close();
  } catch (err) {
    console.error("Failed to send message:", err);
  }
}

sendMessage("Hello, world!");

Here is the output of the code:

Additional info

So far, we have completed this tutorial on message queueing and its role in facilitating seamless communication across various systems. To further improve your skillset, here are some additional best practices that should be implemented when building complex services:

• Rate limiting

• Load balancing

• Application monitoring and logging

• Continuous integration and deployment

Summary

We've highlighted the importance of message brokers and how to implement message queueing in a backend application.

Feel free to check out my other articles here. Till next time, keep on coding!

In this article, we will discuss some measures you can take to make systems much more efficient and reliable.

What is a Distributed System?

In a very simple way, a distributed system is a group or collection of independent computer nodes connected via a central unit which share and utilise computational resources.

Think of this as a network of systems which can interact and share tasks which they can process collectively as a single unit. Therefore distributed workloads refer to the processes or tasks that run in these systems.

Now, for these systems to interact efficiently, you can make use of some protocols or designs. One of them is a queue, which you'll learn about in the next section.

What is a Queue?

A queue is a data structure which models the First-In-first-Out Principle (FIFO).

To understand this, picture a checkout station in a supermarket. The first person to join the line will be the first to be attended to. Then subsequently, the next person is attended to. This is pretty much how queues works, but now as a data structure what is being processed can be different events or messages which we will identify as workloads.

Typically in a queue system, you have the producer and the consumer. The producer is the service that sends an event or in this context, a workload into the queue. And the consumer is the service that listens and processes it.

Streamlining Workflows: Producers Feed the Queue, Consumers Process the Flow.

With the very nature of these systems, running queues across this network can be demanding and sometimes you resort to some challenges which will be discussed briefly.

Challenges of Long-Running Queues in a Distributed System.

The goal of a task queue is primarily to handle asynchronous or long-running tasks efficiently.

However, within the distributed system context, running workloads across a network can sometimes prove to be challenging as they become overwhelming for these queues. This can leads to problems such as the following:

• Increase In Network Latency: Distributed systems primarily rely on communications across networks to efficiently coordinate workloads across the different computer nodes. Delayed or long-running tasks can lead to unpredictable response time(latency).
• Difficulty in tracking and monitoring: It can get very complex very quickly to track down each workload especially now that they are shared across different running systems in the network. In order to efficiently manage these you will have to make use of specialised tools such as the ELK Stack.
• Distributed Coordination: Ensuring that workloads being passed to different queues across the network are executed correctly and in the order they ought to is a challenge. Lack of this coordination further leads to data inconsistency.
• Fault Tolerance and Resilience: Processes running in a distributed system will surely contend with breakpoints and node failures. Having workloads in long-running queues can increase these failure points, which can lead to an increase in queue backlogs, data loss or inconsistency.

The question is how best can you combat these issues, or what measures should you take to reduce the possibility of system failures or data inconsistency. An approach to solving this is by making use of partitioned task queues.

What does it mean to partition a Queue?

Partitioning can be defined as breaking down a component into smaller subsections. Using this, you can derive a definition for queue partition as breaking a queue system into smaller, more manageable bits to offload workloads in a faster and more efficient manner.

Picture a centralised queue with various workloads coming with different task types or even priorities.

When traffic or workloads are at a very high rate, this queue will eventually become overwhelmed and keep compiling backlogs.

In order to lift the burden you can divide the queue into various bits to handle different task types. For example, you can have a queue to handle priority tasks or even to process different event types, like notifications.‌

Partitioned Queues in Distributed Systems: Optimizing Workload Management Across Nodes.

Primarily the goal of partitioning a queue is to improve system performance and reduce downtimes or data inconsistency. In order to do this, you have to look at different strategies of how to partition queues in other to know the one that is best suitable for your system.

Effective Task Queue Partitioning Strategies

There are various ways in which you can break a queue into subsections, but for you to be able to do that efficiently, you can employ some tested strategies such as Range-based Partitioning, Hash-based partitioning, and so on.

Hash-based partitioning

This form of partitioning allows for the selection of a queue using an assigned hash value.

First off, a parameter in the workload is used to get a hash value. Then, you take the modulo of that value with the number of partitions (queues), which results in a partition index. This index is then used to determine the queue to which the task will be assigned.

const numPartitions = 4;

const queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    partitionIndex: i + 1,
}));

function getPartitionIndex(hash) {
    const partitionIndex = hash % numPartitions;
    return partitionIndex + 1;
}

function hashCode(str) {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
        const char = str.charCodeAt(i);
        hash = ((hash << 5) - hash) + char;
    }
    return Math.abs(hash);
}

const tasks = [
    { id: 1, value: "Emails Task" },
    { id: 2, value: "Transactions Tasks"},
    { id: 3, value: "Orders Tasks" },
    { id: 4, value: "Update Tasks" },
];

tasks.forEach(task => {
    const hash = hashCode(task.value);
    const partitionIndex = getPartitionIndex(hash);
    const assignedQueue = queues.find(queue => queue.partitionIndex === partitionIndex);
    console.log(`Task with ID: ${task.id} (content: ${task.value}) assigned to ${assignedQueue ? assignedQueue.name : "no queue"}`);
});

The above example involves hashing a particular value or field in the workload or event that is to be assigned. In this case, the content field was hashed, and its hash value was then used to determine which queue the workload will be assigned to. This hash value might appear random but in truth it is constant, meaning the same input will always produce the same hash value.

Task with ID: 1 (content: Emails Task) assigned to Queue 3
Task with ID: 2 (content: Transactions Tasks) assigned to Queue 4
Task with ID: 3 (content: Orders Tasks) assigned to Queue 4
Task with ID: 4 (content: Update Tasks) assigned to Queue 2

Seeing the output, if you were to change the content of one of these tasks, you will see that they will get another hash value and might be assigned to a different queue.

In a real-life scenario, if you want the workloads to be distributed amongst different queues, you should ensure the hash value is something unique among tasks, like its Identifier. You can make use of an UUID.

Range-based Partitioning

This strategy allows you to spread tasks or workloads coming from a centralised workload into subsections based on a range of task IDs. For example, queue A might only allow a task ID that falls within the range of 1 - 50, while queue B handles the task with an ID ranging from 51- 100.‌

‌This strategy ensures that tasks are distributed across each partition in order to efficiently utilise the resources available and ensure workloads are distributed.

Take a look at this code example below that shows how tasks are assigned to a partition based on the range their IDs fall in:

const { v4: uuidv4 } = require("uuid");

const numPartitions = 4;
const partitionRanges = Array.from({ length: numPartitions }, (_, i) => ({
    start: i * 250,
    end: (i + 1) * 250,
}));

const queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    partitionIndex: i + 1,
}));

function getPartitionIndex(hash) {
    return Math.floor(hash / 250) + 1;
}

const hashCode = (str) => {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
        const char = str.charCodeAt(i);
        hash = (hash << 5) - hash + char;
    }
    return Math.abs(hash) % 1000;
};

const tasks = Array.from({ length: 4 }, () => ({ id: uuidv4() }));

tasks.forEach((task) => {
    const hash = Math.abs(hashCode(task.id));
    const partitionIndex = getPartitionIndex(hash);
    const assignedQueue = queues.find(
        (queue) => queue.partitionIndex === partitionIndex,
    );
    console.log(
        `Task with ID: ${task.id} assigned to ${assignedQueue ? assignedQueue.name : "no queue"}`,
    );
});

In this code example, the taskId is a UUID. To match them to a partitioned queue range (each queue has a range of 250), you first hash-code it and then convert it to an absolute value.

Next, you call the PartitionIndex() function with this hash value.

From there, It finds the partitioned queue that the taskId fits within the range.

This is an example output below:

Task with ID: 493fbf04-2f4f-43c0-9486-f5c99313d4e6 assigned to Queue 2
Task with ID: 9bad272f-3369-408e-bb37-5ef00c68b0b6 assigned to Queue 3
Task with ID: e0ec2d06-d66b-4e0e-8e85-04e4755a42be assigned to Queue 1
Task with ID: 960f415b-2e64-4169-b4ee-59b11c33b451 assigned to Queue 4

Round-Robin Partitioning

For this partitioning strategy, tasks are assigned to queues in a cyclic pattern. This means that if a task gets assigned to queue 1, the next task will get assigned to queue 2, until all the queues get assigned a task before starting from queue 1 again, therefore making it to be distributed evenly amongst the queues.

const numPartitions = 4;
let currentPartition = 0;

function getRoundRobinPartitionIndex() {
    const partitionIndex = (currentPartition % numPartitions) + 1;
    currentPartition++;
    return partitionIndex;
}

const tasks = [1, 2, 3, 4, 5]; 
tasks.forEach(task => {
    const partitionIndex = getRoundRobinPartitionIndex();
    console.log(`Task with ID: ${task} assigned to Queue ${partitionIndex}`);
});

For this example, a function getRoundRobinPartitionIndex() was defined and it determines the next queue index using a round-robin strategy.

It first calculates the queue index by taking the modular value of currentPartition and numPartitions. It adds 1 to it to ensure the initial index starts from 1, then increments the currentPartition to make sure the next queue to be assigned is greater than the previous. If the modular value is 0, it returns to the queue with index of 1.

Task with ID: 1 assigned to Queue 1
Task with ID: 2 assigned to Queue 2
Task with ID: 3 assigned to Queue 3
Task with ID: 4 assigned to Queue 4
Task with ID: 5 assigned to Queue 1

As you can see, the output above shows that the tasks were distributed to each of the queues and then started from the first queue again.

Dynamic Partitioning

This type of partitioning strategy involves assigning tasks to queues based on changing workloads or system conditions.

For instance, a system has 3 queues actively working and 1 that is free. It checks for the one with the least tasks assigned to it and then assigns a new task to it. By doing this, it ensures that the queues are not overwhelmed and that system resources are efficiently utilised.

const { v4: uuidv4 } = require('uuid');
let numPartitions = 4;
let queues = Array.from({ length: numPartitions }, (_, i) => ({
    name: `Queue ${i + 1}`,
    tasks: [],
}));

function dynamicPartitioning(task) {
    let minLoadQueue = queues[0];
    for (let queue of queues) {
        if (queue.tasks.length < minLoadQueue.tasks.length) {
            minLoadQueue = queue;
        }
    }
    minLoadQueue.tasks.push(task);
    console.log(`Task with ID: ${task.id} assigned to ${minLoadQueue.name}`);
}

const tasks = Array.from({ length: 5 }, () => ({ id: uuidv4(), value: `Task content` }));

tasks.forEach(task => {
    dynamicPartitioning(task);
});

function addPartition() {
    numPartitions++;
    queues.push({ name: `Queue ${numPartitions}`, tasks: [] });
    console.log(`Added new partition: Queue ${numPartitions}`);
}

setTimeout(() => {
    addPartition();
    // Assign new tasks to demonstrate dynamic partitioning with the new queue
    const newTasks = Array.from({ length: 5 }, () => ({ id: uuidv4(), value: `New task content` }));
    newTasks.forEach(task => {
        dynamicPartitioning(task);
    });
}, 5000);

In this code above, it starts by creating three queues and dynamically assign tasks to the least loaded queue.

After a simulated put-off, a new queue is brought to deal with the elevated workload and new obligations are distributed among all available queues.

This technique ensures that the burden is balanced throughout all queues, and the queue is overloaded, making the machine more efficient and responsive to adjustments in workload.

Task with ID: 396d4fde-830b-4443-b59b-c089f3a1db49 assigned to Queue 1
Task with ID: aa85af19-e46a-4b13-9a9c-2eb3ec535af0 assigned to Queue 2
Task with ID: f93d4cbb-0cad-4b71-a424-4c27a77ed38a assigned to Queue 3
Task with ID: c7e0cc7e-f055-4a38-9869-466406a33ca8 assigned to Queue 4
Task with ID: b49fe153-73e8-45d1-85b3-a5873f836dc9 assigned to Queue 1
Added new partition: Queue 5
Task with ID: a8a5f78e-da7e-4c65-9d74-8cef57f271f3 assigned to Queue 5
Task with ID: 00ce47b0-6098-49d6-ad06-820733ffe16e assigned to Queue 2
Task with ID: 7b1a5dcf-c42c-46dc-8e29-bae1cc5408c1 assigned to Queue 3
Task with ID: 3e31ad34-1a19-4d3c-a8f3-08f765a45f13 assigned to Queue 4
Task with ID: 73d55243-27da-44a4-9192-532a20889373 assigned to Queue 5

There are other strategies, like content-based partitioning, random-based partitioning, or even a user-defined custom type.

All these strategies are good but can still have some issues if not properly managed.

For instance, if you are using the range-based partition strategy and you select an identifier which falls within the set range of a queue, that queue partition will eventually get overwhelmed. It is left up to you to figure out the one that is best for your system requirements.

Conclusion

Processing workloads in distributed systems can be kind of complex and inefficient if not managed properly.

Partitioning queues across the system and having the workloads be distributed to them based on a strategy is one way to improve your system performance.

In this article, some challenges with long running queues were pointed out, and some strategies to relieve them were given. I have been researching distributed systems in run-time environments and databases and this was one of my findings.

I hope you enjoyed reading this because I enjoyed writing this. You can follow me on Twitter(X).

So you might wonder – how do you make sure your application runs without breaking? To handle this, you can send tasks to a queue outside your application's flow.

What is a Message Queue?

A message queue is a tool that facilitates the communication and transfer of data between services within a single application (or externally). It stores these data or messages using the First-In-First-Out (FIFO) principle. This means that older data that's passed into these queues gets processed before newer data.

Different components make up a message queue, such as:

• Messages: These are the data that are sent to the queue. They are often referred to as jobs.
• Queues: These are the data structures used for storing messages.
• Producers: These are a service that sends messages or data into the queue system.
• Consumers: These are a service that listens to the queue and executes messages passed in it.

Message Queuing Tools

Now, there are various message queuing tools you can use in asynchronous systems, like the following:

• RabbitMQ: a reliable and flexible option for implementing message queues in applications.
• Apache Kafka: an efficient message queuing tool, also very good at event stream processing.
• Redis: an in-memory store used for message queuing, caching, and data processing.

Note that some of these tools are not limited to message queuing but can be used for other purposes as well, like stream processing.

In this article, you will create a simple Nest.js project which will use RabbitMQ as the Message Queue Service Provider.

The tutorial will be divided into 3 parts:

• How to Set Up a Nest.js Project for Basic User Registration Flow
• How to Set Up an Email Service for User Registration
• How to Integrate a Queue Service using RabbitMQ

Prerequisites

• You'll need to have Node installed on your system. If you don't have it, here is its official site: https://nodejs.org/en.
• You'll need to have Node Package Manager (NPM) installed, which you can download here if you don't have it: https://docs.npmjs.com/downloading-and-installing-node-js-and-npm.
• You'll need to have installed RabbitMQ. Here is where you can get it in case you haven't yet: https://www.rabbitmq.com/download.html
• You'll need to have a text editor. For this article, I'll use VSCode. You can download it here: https://code.visualstudio.com/download or use the code editor of your choice.

How to Set Up a Nest.js Project

Spinning up a Nest.js application is fast and simple if you use the Nest CLI. Open up your terminal and enter this command below to install the CLI:

 $ npm install -g @nestjs/cli

This installs the Nest.js CLI globally on your system, meaning you can call the CLI commands regardless of the directory you are currently in.

Moving forward, to create a simple REST API project, you will enter the command below:

nest new simple-queue

Simple-queue here is the directory name that will be created. Inputting this command gives you a prompt to select a package manager.

When that's done, navigate to the created directory and open it in your text editor by entering this command:

cd simple-queue && code .

This opens up your text editor. We want to work on a project that will best show how a message queue can be used in a real-world scenario – so let's set up a basic user registration form. On successful data entry it sends an email to the user, but you'll handle the email service separately by passing it into a queue to improve performance.

For this, we'll be using an SQLite database, TypeOrm, class-validators, and the dotenv package so you can secure your config variables. Go ahead to install them by typing this command in your terminal:

npm install --save @nestjs/typeorm typeorm sqlite3 class-validator dotenv

When the installation is complete, go to your root app module, and then include the TypeOrm configuration for your database.

SQLite is a lightweight SQL database which allows us to quickly spin up and test data. It's optimal for this use case – and now we'll configure it.

Configuring the SQLite Database

import { Module } from "@nestjs/common";
import { TypeOrmModule } from "@nestjs/typeorm";
import { AppController } from "./app.controller";
import { AppService } from "./app.service";

@Module({
  imports: [
    TypeOrmModule.forRoot({
      type:'sqlite',
      database: 'mini-db.sqlite',
      entities: [__dirname + '/**/*.entity{.ts,.js}'],
      synchronize: true,  
  })],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Congrats! You have successfully connected a DB to your project. Now it's time to create the services that will handle the user registration.

In order to do this, you will have to go back to your dear friend the Nest CLI. There, you'll be inputting a different command to help generate a resource folder for the User, which will contain the entity, service, dto, and the controller.

To do this, open your terminal and enter in this command:

nest generate resource users

A prompt to select your transport layer will be shown. Select the first one which is the REST API. Then, another prompt will ask if you would like to generate CRUD endpoints – you can type Yes. Then you can make modifications according to your requirements.

To proceed, you first have to define what information each user should have. First, create a User entity. You can do this by navigating to the user entity file in the created entity subfolder in the user folder. Then define the user data like this:

import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";

@Entity('users')
export class User {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ length: 100, unique: true })
  username: string;

  @Column({ length: 100, unique: true })
  email: string;
}

For this mini-project, you'll use basic user data to make the process faster. The username and email field have been set to be unique, meaning that there won't be a duplicate of the data instance passed in for this user table.

Now having done this, modify the create user dto file that was generated like this:

import { IsNotEmpty, IsString, IsEmail } from "class-validator";

export class CreateUserDto {
    @IsNotEmpty()
    @IsString()
    username: string;

    @IsNotEmpty()
    @IsString()
    @IsEmail()
    email: string;
  }

This was created to validate the payload that will be sent in your request by using the class-validator package.

Now, modify the create method in the user service file.

import { Injectable } from "@nestjs/common";
import { InjectRepository } from "@nestjs/typeorm";
import { Repository } from "typeorm";
import { CreateUserDto } from "./dto/create-user.dto";
import { User } from "./entities/user.entity";

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    return await this.userRepository.save(newUser);
  }
}

Next you'll modify the controller file. You've already defined the create endpoint, so you'll just have to clean up the other endpoints that are not needed.

import { Controller, Post, Body } from "@nestjs/common";
import { CreateUserDto } from "./dto/create-user.dto";
import { UsersService } from "./users.service";

@Controller('users')
export class UsersController {
 constructor(private readonly usersService: UsersService) {}
  @Post()
  create(@Body() createUserDto: CreateUserDto) {
    return this.usersService.create(createUserDto);
  }
}

Open up the user module file and make some adjustments by adding the import field to the Module decorator and using the TypeOrmModule property.

import { Module } from '@nestjs/common';
import { UsersService } from './users.service';
import { UsersController } from './users.controller';
import { TypeOrmModule } from '@nestjs/typeorm';
import { User } from './entities/user.entity';

@Module({
  imports: [TypeOrmModule.forFeature([User])],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Next, start up your server by entering this command on your terminal: npm run start:dev. Once the server is up and running, open up your API client of choice. For this article, we'll use Postman. Then make a POST request to the endpoint, which will be localhost:3000/users, providing the payload data required.

A request was made and a user instance was created.

Next up is adding an email service to your project which will help notify new users who are registering.

How to Set Up an Email Service for User Registration

For this, you'll use some packages which are required to create an email service. Open up your terminal and input the command below to install these packages:

npm install --save @nestjs-modules/mailer nodemailer

When these packages are installed, you can now implement the mail service. Using the Nest CLI, create a mailer module and service by entering this command in your terminal:

nest generate module email && nest generate service email

When it's done, open up the newly created module file in the mail folder. You'll use the MailerModule property of the @nestjs-modules/mailer package to configure your mail service here. It requires an SMTP client whose keys you'll need to configure this MailerModule.

For that you can use https://app.elasticemail.com to get these SMTP keys. Sign up and connect to the SMTP API. You'll then be given keys for your private use.

Note that this free mode of the SMTP client has limitations and it cannot send to all emails – so you should use a test email service.

How to Configure the Mailer Module

Once you have gotten that set up, go back to your application and create a .env file. Set your secrets for the SMTP keys. Then configure your MailerModule like this:

import { Global, Module } from "@nestjs/common";
import { EmailService } from "./email.service";
import { MailerModule } from "@nestjs-modules/mailer";

require('dotenv').config();
@Global()
@Module({
  imports: [
    MailerModule.forRoot({
      transport: {
        service: 'QueueTest',
        host: process.env.SMTP_HOST,
        port: process.env.SMTP_PORT,
        auth: {
          user: process.env.SMTP_USER,
          pass: process.env.SMTP_PASSWORD,
        },
      },
      defaults: {
        from: process.env.FROM_EMAIL,
      },
    }),
  ],
  providers: [EmailService]
})
export class EmailModule {}

The global decorator was set in order to make sure the MailModule can be called anywhere in your application. Make sure your secrets are properly loaded and thaqt you have a valid Email set in the from: process.env.FROM_EMAIL.

Check to make sure that the EmailModule is also imported in the root App Module the same way your UsersModule was imported in the Imports Array of the App Module.

Next, open your email service file – you'll need to make some modifications to the EmailService class. Add a constructor and call the MailService property from the @nestjs-modules/mailer package. Then go ahead and create a function that will handle sending the emails.

Below is a class and method that does this:

import { MailerService } from '@nestjs-modules/mailer';
import { HttpException, HttpStatus, Injectable } from '@nestjs/common';

@Injectable()
export class EmailService {
  constructor(private mailerService: MailerService) {}
  async sendEmail(options: { email: string; subject: string; html: string;
  }) {
    try {
      const message = {
        to: options.email,
        subject: options.subject,
        html: options.html
      };
      const emailSend = await this.mailerService.sendMail({
        ...message,
      });
      return emailSend;
    } catch (error) {
      throw new HttpException('Error', HttpStatus.INTERNAL_SERVER_ERROR);
    }
  }
}

Now you've defined the method to send an email. You've also put an exception handler in place for better error handling.

Now it's time to add this newly created service to your user registration flow.

Navigate to your user service file, and add the mail service to your constructor as a provider. Then call the service in your create user method like this:

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>,
    private emailService: EmailService
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    const user =  await this.userRepository.save(newUser);
      const emailData = {
        email: user.email,
        subject: 'Welcome to Our Community',
        html: `<p>Hello ${user.username},</p>
        <p>Welcome to our community! Your account is now active.</p>
        <p>Enjoy your time with us!</p>`,
      };
      await this.emailService.sendEmail(emailData)
    return user
  }
}

Make sure to modify your modules in order to correct any dependency injection errors. In your email module file, add the EmailService to the exports array:

 providers: [EmailService],
 exports: [EmailService]

Add it below your providers to export the Email Service so it can be accessed in other modules. Then import the EmailModule to your User module file and add it to your import array like this:

@Module({
  imports: [TypeOrmModule.forFeature([User]), EmailModule],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Now it's time to test it. Get a free account from any online email testing platform and open Postman. Make a request to the create user endpoint with your valid email. You should get an email response like this:

Email response you should get

How to Integrate a Queue Service using RabbitMQ

To get started with this, you'll have to install some packages that let you implement queues using RabbitMQ. Enter the command below to install these packages:

npm install --save amqplib @types/amqplib amqp-connection-manager

Configure the Producer Service

Once installation is complete, it's time to configure RabbitMQ. You'll create a new folder in your src directory and name it queues. Then create the queue producer file. Import these packages and set them up like this:

import { HttpException, HttpStatus, Injectable, Logger } from '@nestjs/common';
import amqp, { ChannelWrapper } from 'amqp-connection-manager';
import { Channel } from 'amqplib';

@Injectable()
export class ProducerService {
  private channelWrapper: ChannelWrapper;
  constructor() {
    const connection = amqp.connect(['amqp://localhost']);
    this.channelWrapper = connection.createChannel({
      setup: (channel: Channel) => {
        return channel.assertQueue('emailQueue', { durable: true });
      },
    });
  }

  async addToEmailQueue(mail: any) {
    try {
      await this.channelWrapper.sendToQueue(
        'emailQueue',
        Buffer.from(JSON.stringify(mail)),
        {
          persistent: true,
        },
      );
      Logger.log('Sent To Queue');
    } catch (error) {
      throw new HttpException(
        'Error adding mail to queue',
        HttpStatus.INTERNAL_SERVER_ERROR,
      );
    }
  }
}

The AMQP connection was set and is running on localhost with the default RabittMQ port which is 5432. You also established a channel on that connection with an option input which is executed anytime a new channel is created. This helps if you have any configuration for that channel.

You also created an emailQueue with the assertQueue property which checks that a queue with that name does not already exist. If it does exist, it has no effect so it's idempotent.

Then you created an option durable: true to make sure that the queue will survive a server restart.

Next, you defined the method to add the email data to a queue. This calls the sendToQueue property of the channelWrapper, passing in the queue name you want to send the data to. Ideally, it should be the same name as the one you defined with the assertQueue property.

The second argument is the mail data, but firstly you converted it to a JSON string then to a Buffer. You do this because messages in RabbitMQ are mostly transmitted as binary data.

You can then set an option persistent: true to ensure that the data being sent to the queue won't be lost if the server crashes. Then with some error handling and the method to send messages to the queue, it's good to go.

Set Up the Consumer Service

Now that you've configured the producer service, it's time to set up the consumer service.

Create another file in the queue sub-folder. It's quite similar, but in this case, you will be consuming the data from the queue. Below is the configuration for the consumer service:

import { Injectable, OnModuleInit, Logger } from '@nestjs/common';
import amqp, { ChannelWrapper } from 'amqp-connection-manager';
import { ConfirmChannel } from 'amqplib';
import { EmailService } from 'src/email/email.service';

@Injectable()
export class ConsumerService implements OnModuleInit {
  private channelWrapper: ChannelWrapper;
  private readonly logger = new Logger(ConsumerService.name);
  constructor(private emailService: EmailService) {
    const connection = amqp.connect(['amqp://localhost']);
    this.channelWrapper = connection.createChannel();
  }

  public async onModuleInit() {
    try {
      await this.channelWrapper.addSetup(async (channel: ConfirmChannel) => {
        await channel.assertQueue('emailQueue', { durable: true });
        await channel.consume('emailQueue', async (message) => {
          if (message) {
            const content = JSON.parse(message.content.toString());
            this.logger.log('Received message:', content);
            await this.emailService.sendEmail(content);
            channel.ack(message);
          }
        });
      });
      this.logger.log('Consumer service started and listening for messages.');
    } catch (err) {
      this.logger.error('Error starting the consumer:', err);
    }
  }
}

First, you defined your consumer class. For this, it implements the onModuleInit interface which is provided by @nestJs/common. This specifies that the defined class should have a method named onModuleInit().

Like the name says, the method will be called automatically during the module initialization which is when the module containing this class is loaded.

In the class constructor, you added the emailService because you'll be using the sendEmail method of that class.

In the onModuleInit() method, you defined a channel. This is necessary because you need a channel to consume messages from a queue.

From this, the channel is then used to assert a queue which should be similar in name and options to what you have on your producer service. If it's not, you won't be able to listen to the queue created on the producer service.

Then you used the consume method of channel to listen and execute the message coming from the queue you have registered.

Recall that before, you had to convert the message to Buffer in order to send it into a queue. Now, you have to convert it to a JavaScript object. Then call the emailService method to send an email and pass in the converted JavaScript object as the argument of that method.

Finally, you called the ack method which is used to inform the queue that the message has been received and processed successfully in order for it to be removed from the queue.

Now that you've defined these services, create a module file and set them in the providers array. Then export the producer service because you will be calling it in another module.

import { Module } from '@nestjs/common';
import { ConsumerService } from './consumer.file';
import { ProducerService } from './producer.file';

@Module({
  providers: [ProducerService, ConsumerService],
  exports: [ProducerService],
})
export class QueueModule {}

Next up is to add the emails being sent on user registration to the queue service that you just created.

Navigate back to your user service file and make some modifications: replace the email service with the producer service as a provider in the constructor, and then call the service and the method to add to the email queue as shown below:

import { Injectable } from "@nestjs/common";
import { InjectRepository } from "@nestjs/typeorm";
import { Repository } from "typeorm";
import { CreateUserDto } from "./dto/create-user.dto";
import { User } from "./entities/user.entity";
import { ProducerService } from "src/queues/producer.file";

@Injectable()
export class UsersService {
  constructor(
    @InjectRepository(User)
    private readonly userRepository: Repository<User>,
    private producerService: ProducerService,
  ) {}
  async create(createUserDto: CreateUserDto): Promise<User> {
    const newUser = this.userRepository.create(createUserDto);
    const user = await this.userRepository.save(newUser);
    const emailData = {
      email: user.email,
      subject: 'Welcome to Our Community',
      html: `<p>Hello ${user.username},</p>
        <p>Welcome to our community! Your account is now active.</p>
        <p>Enjoy your time with us!</p>`,
    };
    await this.producerService.addToEmailQueue(emailData);
    return user;
  }
}

Also in the user module file, replace the EmailModule with that of the QueueModule to avoid dependency injection errors when you start up your server.

import { Module } from '@nestjs/common';
import { UsersService } from './users.service';
import { UsersController } from './users.controller';
import { TypeOrmModule } from '@nestjs/typeorm';
import { User } from './entities/user.entity';
import { QueueModule } from 'src/queues/queue.module';

@Module({
  imports: [TypeOrmModule.forFeature([User]), QueueModule],
  controllers: [UsersController],
  providers: [UsersService],
})
export class UsersModule {}

Now finally, it's time to test the user registration flow again. So navigate back to Postman and then type in a valid email and username and hit enter. On the terminal of your server running, you will see logs that were set in order to track the way the message got sent and how it was received and executed.

Logs that help you track the message

You can also open up the RabbitMQ dashboard to view queue activity on http://localhost:15672, By default the user is "guest", so enter in guest for the username and password.

RabbitMQ Queues and Streams

Here's the link to the GitHub repository. Feel free to check it out whenever you're stuck.

Conclusion

In this article you learned what a message queue is along with some major components of how they work. You also built a mini Nest.js project and implemented an email service in it. Finally, you integrated the queue service into your project, showing how it works in a real-life scenario.

Understanding message queue behaviors and patterns is an essential skill when developing scalable applications. This helps reduce lag and improves the speed and efficiency of your applications.

I hope you enjoyed reading this article. You can follow me on Twitter.

Message queues help solve this problem by providing an additional branch to the usual request-response journey. This additional branch helps make sure users can get immediate responses, and the time-consuming processes can be done on the side. Everybody goes home happy.

This article will focus on explaining what message queues are and how to get started with them by building a very simple application. You should be familiar with the basics of Node.js, and you should have Redis installed either locally or on a cloud instance. Learn how to install Redis here.

What is a Queue?

A queue is a data structure that allows you store entities in an order. Queues use a first-in-first-out (FIFO) principle.

The concept of queues in computer science is the same as the concept of queues in everyday life where people line up to get things. You join a queue from the back, wait till it is your turn, then leave the queue from the front after you have been attended to.

In computer science, when a process like an API request is running, and you need to remove a certain task (like sending an email) from the current flow, you push it to a queue and continue the process.

The diagram below illustrates the lifecycle of a queue:

Queue Lifecycle | https://optimalbits.github.io/bull/

What is a Job?

A job is any piece of data that is used on a queue, usually a JSON-like object.

As demonstrated in the cover image of this article, you can think of a job as each person on a queue at an airport. Each person holds a briefcase containing specific data, and other instructions (passports and maybe medical papers where required) that will help when it is their turn to be attended to.

New people joining this queue will enter from the back (as the last person), and people will be attended to from the front. That is how jobs are also processed, each job contains data that will be used for its processing. New jobs are added from the back while jobs are taken out from the front.

What is a Job Producer?

A job producer is any piece of code that adds a job to a queue. In real life, this would be the security guard at the airport that gives direction to people, telling them which queue to join for different purposes.

A job producer can exist independently of a job consumer. This means that in a microservice setup, a particular service might just be concerned with adding jobs to a queue, but not how they're processed after.

What is a Worker (Job Consumer)?

A worker, or job consumer, is a process, or function, that can execute a job. Think of a worker as a bank cashier attending to people on a queue at the bank. When the first person comes in, they join the queue as the only one on the queue. The cashier then calls for them and the queue is emptied.

The cashier requests for specific details to be used to process the transaction from the person. While the cashier attends to that customer, four more customers could have lined up. They will remain on the queue till the cashier is done with the first customer before calling for the next one. This is the same process with queue workers — they pick the first job in the queue, and process it.

What are Failed Jobs?

Often times, some jobs might fail during processing.

Here are some reasons why a job could fail:

• Invalid or missing input data: When data required for a job to be processed is missing, the job will fail. For example, a job to send an email will fail without the recipient's email address.
• Timeout: A job could be failed by the queue mechanism if it is taking longer than usual. This could be due to an issue on a dependency of the job or something else, but usually you don't want a single job running forever.
• Network or infrastructure problems: These problems are almost out of your control, but they do happen. A database connection error for example would fail a job.
• Dependency issues: Sometimes a job relies on some external resources to function well. Whenever these other resources are unavailable or unsuccessful, the job will fail.

When jobs fail, you can configure your queue mechanism to retry them. You can either retry the job immediately, or after a calculated amount of time. You can set a maximum number of attempts, which is recommended. If not, you end up running a job that will always fail infinitely.

Why Use Queues?

Queues are useful for creating robust communication channels between microservices. Multiple services can use the same queue. Different services could be tasked with different problems. When a service completes its task, it can push a job to another service that has workers waiting for that job. That service will pick it up and do whatever is needed with the data.

Queues are also useful for offloading heavy tasks from a process. As you'll see in this article, a time consuming task like sending of an email can be put on a queue to avoid slowing down response time.

Queues help avoid single points of failure. A process that has the ability to fail and can be retried is best processed using a queue where it can be retried after a while.

How to Build a Simple Application that Uses Queues

In this article, we'll build a simple project using Node.js and Redis. We'll use the Bull library as it simplifies a lot of the complexities involved in building a queue system. The project will have a single endpoint to send emails.

Create a New Node.js Project

mkdir nodejs-queue-project
cd nodejs-queue-project
npm init -y

The commands above will create a new folder named nodejs-queue-project and a package.json file in it. The package.json file should look like this:

{
  "name": "nodejs-queue-project",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "keywords": [],
  "author": "",
  "license": "ISC"
}

Install the Required Dependencies

npm i express @types/express @types/node body-parser ts-node ts-lint typescript nodemon nodemailer @types/nodemailer

The commands above will install the different packages and dependencies required for the project.

After installation, you can update the scripts section of your package.json to have a dev command. Your whole package.json file should look like this now:

{
  "name": "nodejs-queue-project",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "dev": "nodemon src/app.ts"
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "dependencies": {
    "@types/express": "^4.17.17",
    "@types/node": "^20.3.3",
    "@types/nodemailer": "^6.4.8",
    "body-parser": "^1.20.2",
    "express": "^4.18.2",
    "nodemailer": "^6.9.3",
    "nodemon": "^2.0.22",
    "ts-lint": "^4.5.1",
    "ts-node": "^10.9.1",
    "typescript": "^5.1.6"
  }
}

The file above shows all your installed dependencies. The npm run dev command will run when you use the dev script.

How to Build the Endpoint

The first thing to do is to create a new folder named src. This folder will contain all your code files. The first file it will contain is the root file of the application — the app.ts file as defined in the package.json file.

We'll use the app.ts file to import required packages and create a simple server with a single endpoint to send a email as seen below:

import express from "express";
import bodyParser from "body-parser";
import nodemailer from "nodemailer";

const app = express();

app.use(bodyParser.json());

app.post("/send-email", async (req, res) => {
  const { from, to, subject, text } = req.body;

  // Use a test account as this is a tutorial
  const testAccount = await nodemailer.createTestAccount();

  const transporter = nodemailer.createTransport({
    host: "smtp.ethereal.email",
    port: 587,
    secure: false,
    auth: {
      user: testAccount.user,
      pass: testAccount.pass,
    },
    tls: {
      rejectUnauthorized: false,
    },
  });

  console.log("Sending mail to %s", to);

  let info = await transporter.sendMail({
    from,
    to,
    subject,
    text,
    html: `<strong>${text}</strong>`,
  });

  console.log("Message sent: %s", info.messageId);
  console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));

  res.json({
    message: "Email Sent",
  });
});

app.listen(4300, () => {
  console.log("Server started at http://localhost:4300");
});

Now, you can start your server by running npm run dev in your terminal. You should see a message saying Server started at [http://localhost:4300](http://localhost:4300) in your terminal.

npm run dev message

You can now test the endpoint using a tool like Postman:

Endpoint testing on Postman

The request took almost 4 seconds as shown in the screenshot. This is very slow for an endpoint. If you take a look at your terminal, you should also see a URL where you can preview the email that was sent.

Opening the link lets you see how the email looks.

Email content

How to Create the Queue

To make the process even faster, the email can be queued to be sent later and a response sent to the user immediately.

To do this, install the bull library and its @types library as we'll use it to create a queue. That is:

npm i bull @types/bull

Creating a new queue using bull is as easy as instantiating a new Bull object with a name for the queue:

// This goes at the top of your file
import Bull from 'bull';

const emailQueue = new Bull("email");

When the queue is created with just the queue name, it tries to use the default Redis connection URL: localhost:6379. If you prefer using a different URL, simply pass in a second object to the Bull class as an options object:

const emailQueue = new Bull("email", {
  redis: "localhost:6379",
});

At this point, you can create a simple function to serve as a job producer and add a job to the queue every time a request comes in.

type EmailType = {
  from: string;
  to: string;
  subject: string;
  text: string;
};

const sendNewEmail = async (email: EmailType) => {
  emailQueue.add({ ...email });
};

This newly created function, sendNewEmail, accepts an object containing details of the new email to be sent of type EmailType. There's sender email address (from), recipient email address (to), subject of the email, and the content of the email (text). Then it pushes a new job to the queue.

You can use this function instead of sending the email during the request now. Modify the endpoint to do this:

app.post("/send-email", async (req, res) => {
  const { from, to, subject, text } = req.body;

  await sendNewEmail({ from, to, subject, text });

  console.log("Added to queue");

  res.json({
    message: "Email Sent",
  });
});

At this point, the code is simpler and the process is faster. The request only takes about 40m — about 100x faster than before.

Endpoint testing with Postman

At this point, the email is added to a queue. It will remain on the queue until processed. The job can be processed by the same application or another service (if in a microservice setup).

How to Process the Jobs

The cycle is incomplete and useless if the mails never leave the queue. We'll create a job consumer to process the jobs and clear the queue.

We can do this by creating the logic for a function that accepts a Job object and sends the email:

const processEmailQueue = async (job: Job) => {
  // Use a test account as this is a tutorial
  const testAccount = await nodemailer.createTestAccount();

  const transporter = nodemailer.createTransport({
    host: "smtp.ethereal.email",
    port: 587,
    secure: false,
    auth: {
      user: testAccount.user,
      pass: testAccount.pass,
    },
    tls: {
      rejectUnauthorized: false,
    },
  });

  const { from, to, subject, text } = job.data;

  console.log("Sending mail to %s", to);

  let info = await transporter.sendMail({
    from,
    to,
    subject,
    text,
    html: `<strong>${text}</strong>`,
  });

  console.log("Message sent: %s", info.messageId);
  console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));

  return nodemailer.getTestMessageUrl(info);
};

The function above accepts a Job object. The object has useful properties that shows the status of and data in a job. Here, we use the data property.

At this point, all we have is a function. It doesn't pick up jobs automatically because it doesn't know which queue to work with.

Before connecting it to the queue, you can go on to add a few jobs to the queue by sending some requests. You can check the email jobs currently queued by running this command in your redis-cli:

LRANGE bull:email:wait 0 -1

This checks the email waitlist, and returns the ids of the waiting jobs.

Redis CLI

I have created a few jobs just to show how workers actually work.

Now, connect the worker to the queue by adding this line of code:

emailQueue.process(processEmailQueue);

This is what your app.ts file should now look after that:

import express from "express";
import bodyParser from "body-parser";
import nodemailer from "nodemailer";
import Bull, { Job } from "bull";

const app = express();

app.use(bodyParser.json());

const emailQueue = new Bull("email", {
  redis: "localhost:6379",
});

type EmailType = {
  from: string;
  to: string;
  subject: string;
  text: string;
};

const sendNewEmail = async (email: EmailType) => {
  emailQueue.add({ ...email });
};

const processEmailQueue = async (job: Job) => {
  // Use a test account as this is a tutorial
  const testAccount = await nodemailer.createTestAccount();

  const transporter = nodemailer.createTransport({
    host: "smtp.ethereal.email",
    port: 587,
    secure: false,
    auth: {
      user: testAccount.user,
      pass: testAccount.pass,
    },
    tls: {
      rejectUnauthorized: false,
    },
  });

  const { from, to, subject, text } = job.data;

  console.log("Sending mail to %s", to);

  let info = await transporter.sendMail({
    from,
    to,
    subject,
    text,
    html: `<strong>${text}</strong>`,
  });

  console.log("Message sent: %s", info.messageId);
  console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));
};

emailQueue.process(processEmailQueue);

app.post("/send-email", async (req, res) => {
  const { from, to, subject, text } = req.body;

  await sendNewEmail({ from, to, subject, text });

  console.log("Added to queue");

  res.json({
    message: "Email Sent",
  });
});

app.listen(4300, () => {
  console.log("Server started at http://localhost:4300");
});

Once you save, you'll notice that the server restarts and immediately starts sending out mails. This is because the worker sees the queue and begins processing immediately.

Server sending out queued emails

Now, both the producer and the worker are active. Every new API request will be pushed to the queue, and the worker will immediately process it unless there's some pending jobs already.

Summary

I hope this article helped you understand what a message queue is, how to add jobs and create processes to run them, and how you can use them to build better web applications. You can find the code files used in this article on GitHub.

If you have any questions or relevant advice, please get in touch with me to share them.

To read more of my articles or follow my work, you can connect with me on LinkedIn, Twitter, and Github. It’s quick, it’s easy, and it’s free!

What Is a Queue?

A queue is linear data structure that consists of a collection is of items that follow a first-in-first-out sequence. This implies that the first item to be inserted will be the first to be removed. You can also say that items are removed in the order they were inserted.

Using a real world example, we can compare a queue data structure to a queue of individuals standing in line for a service. Once one individual is attended to, they leave the queue for the next person to be attended to. They are helped in the order which they came.

Structure of a Queue

A queue is mainly made up of two parts: the front/head and the rear/tail/back. For the purpose of clarity and consistency, we will stick to using front and back.

The back is where the items are inserted and the front is the part of the queue where items are removed/deleted.

Here is a diagram to help you understand better:

The image shows an array with various cells. The items are inserted through the back and removed through the front. There are terms used for the insertion and deletion of items in a queue which we will cover in the next section.

Note that you can reverse the structure of your queue – you can have the front on the right and the back on the left side. Whichever structure you go with, always remember that insertion of items happens through the back and deletion through the front.

Common Operations of a Queue

The following operations are commonly used in a queue:

• Enqueue: Adds an item from the back of the queue.
• Dequeue: Removes an item from the front of the queue.
• Front/Peek: Returns the value of the item in front of the queue without dequeuing (removing) the item.
• IsEmpty: Checks if the queue is empty.
• IsFull: Checks if the queue is full.
• Display: Prints all the items in the queue.

Before we see how to implement this with code, you need to understand how the enqueue and dequeue operations work and how they affect the front and back positions.

The indices of arrays in most programming languages start from 0. While implementing our code, we are going to set the index of the front and back values of our array to -1. This will enable us move the front and back position properly as values are added.

Consider the image below:

The arrows show the position of the front and back of our array. When both positions are at -1, it means the array is empty.

Let us add some items into our array and see what happens.

We have inserted (enqueued) our first item – 5. The position of the front and back have moved as well. Next, we will see what happens as we enqueue more items

A second item has been added but only the back moved. This will continue as we enqueue more items. The front and back moved together in the last example so that the front could assume the position of the first item.

Since that was the first and only item then, the front and back sat at that position. But now that we have enqueued more items, the back will keep following the last item.

We will go on and fill up the array so we can see what happens when we dequeue.

So the back arrow followed the items in the order they were added all the way to the last. Now let's delete (dequeue) some items.

Remember the first-come-first-out sequence? When we execute the dequeue operation, it will first remove 5 from the queue. If we execute it again then it will move to the next number which is 10 and continue in that order for as long as we call it.

Here, the first dequeue operation:

Now the front arrow has moved to index 1. This implies that the item at index 0 has been removed. By removed, we do not mean from the array but from the queue – only items from the front position to the back position are part of the queue.

In the same order, if we keep removing items, it will get to a point where the front arrow meets the back arrow at the end of the queue. If we dequeue again at that point, the front arrow will move past the back arrow and then the queue will be considered empty because there is nothing there to delete. When this happens, we will reset their index to -1 (their initial starting point).

Time for some code!

Queue Implementation in Java

We will break this section down by creating each operation and then putting everything together at the end.

int queueLength = 3;
int items[] = new int[queueLength];
int front = -1; 
int back = -1;

We have created our variables and their parameters. We are using 3 as the maximum number of items that can be enqueued in the array. Like we saw in the images from the previous section, we have set the initial index of the front and back to -1.

Next, we'll define the isEmpty and isFull functionalities.

For isEmpty:

boolean isEmpty(){
      if(front == -1 && back == -1){
          return true;
      } else {
          return false;
      }
  }

Pretty easy to grasp if you followed along in the last section. The array is empty only if the index of front and back is -1.

For isFull:

boolean isFull(){
      if(back == queueLength - 1){
          return true;
      } else {
          return false;
      }
  }

This one might seem a little tricky, but here's the logic: our maximum number of items allowed in the array is 3 but three items in an array is not denoted by index 3 but 2 since the first index is 0. So maximum length minus 1 gives us index 2 which is the third cell in an array.

When all the cells have been enqueued with a value up to the third cell, the array is full.

For enQueue:

void enQueue(int itemValue) {
      if(isFull()){
          System.out.println("Queue is full");
      } else if(front == -1 && back == -1){
          front = back = 0;
          items[back] = itemValue;
      } else{
          back++;
          items[back] = itemValue;
      }
  }

If the array is full then we get a message saying it is full. If the front and back is -1 then the item is assigned to the first cell which is index 0 – otherwise, the value is inserted and the back position is incremented.

For deQueue:

void deQueue(){
      if(isEmpty()){
          System.out.println("Queue is empty. Nothing to dequeue");
      } else if (front == back){
          front = back = -1;
      } else {
          front++;
      }
  }

Here, if the array is empty, we get the corresponding message. If the front has met the back, we reset their index back to -1 like we saw in the images from the previous section. If the last two conditions are not applicable, then the front is incremented.

For display:

void display(){
      int i;

      if(isEmpty()){
          System.out.println("Queue is empty");
      } else {
          for(i = front; i <= back; i++){
              System.out.println(items[i]);
          }
      }
  }

Here, if the array is not empty, we loop through and print all the items.

Lastly, for peek:

void peak(){
      System.out.println("Front value is: " + items[front]);
  }

This simply prints the value of the front item.

These are all the operations for our queue. Here is all of them in one piece below:

// Queue implementation in Java

public class Queue {

  int queueLength = 3;
  int items[] = new int[queueLength];
  int front = -1; 
  int back = -1;

  boolean isFull(){
      if(back == queueLength - 1){
          return true;
      } else {
          return false;
      }
  }

  boolean isEmpty(){
      if(front == -1 && back == -1){
          return true;
      } else {
          return false;
      }
  }



  void enQueue(int itemValue) {
      if(isFull()){
          System.out.println("Queue is full");
      } else if(front == -1 && back == -1){
          front = back = 0;
          items[back] = itemValue;
      } else{
          back++;
          items[back] = itemValue;
      }
  }

  void deQueue(){
      if(isEmpty()){
          System.out.println("Queue is empty. Nothing to dequeue");
      } else if (front == back){
          front = back = -1;
      } else {
          front++;
      }
  }

  void display(){
      int i;

      if(isEmpty()){
          System.out.println("Queue is empty");
      } else {
          for(i = front; i <= back; i++){
              System.out.println(items[i]);
          }
      }
  }

  void peak(){
      System.out.println("Front value is: " + items[front]);
  }

}

Now let us execute the operations:

 public static void main(String[] args) {
    Queue myQueue = new Queue();

    myQueue.enQueue(3);
    myQueue.enQueue(2);
    myQueue.enQueue(1);


    myQueue.display();

    myQueue.peak();


  }

enQueue(3) inserts 3 to our queue, similar to the next two lines of code.

display() prints out the items in the array.

peak() prints the value of the front item.

We did not execute deQueue so you can go on and try it yourself – display your array and take a peek after you dequeue and see what happens. There are various ways to modify the code, so have fun!

Conclusion

In this article, we defined a queue and its structure. We went on to see some examples using images to show how the front and back positions of a queue react when items are enqueued and dequeued.

Lastly, we saw how to implement the queue data structure using arrays in Java.

Thank you for reading and happy coding!

An introduction to RabbitMQ, message broker, AMQP model and more.

In distributed systems, communication between various applications plays an important role. Effectively passing messages between applications was always a crucial decision in system design. One of the elegant solutions to pass messages around your distributed system is a message broker.

In distributed systems, communication between various applications plays an important role. Effectively passing messages between applications was always a crucial decision in system design. One of the elegant solutions to pass messages around your distributed system is a message broker.

They bring in decoupling between the applications and provide an effective way to communicate. With message brokers, an application needs no prior knowledge of their recipients to communicate.

However, what is RabbtiMQ? How does RabbitMQ fit in this picture? Also, what is AMQP?

By the end of this article, we will be able to answer these questions. I also added a few animations so that you can visualize RabbitMQ concepts.

So are you excited? I am! If you ever had difficulties understanding message brokers, as I did, then this article is the right place to start your journey. Stay with me ?

Message Broker

In general, a broker is a person who facilitates trades between a buyer and a seller. An example could be a real estate agent or a stockbroker.

Similarly, if we want to trade messages between two distributed software components, we need a mediator. This mediator is known as the message broker. It receives incoming messages from a sender and sends them to a recipient. This way the sender and receiver can be totally isolated.

Another analogy for a message broker can be a Post Office (see Figure 1). Let’s take a scenario where you are going to send a letter to your cousin living in another city. Then as per this analogy, you are a producer, your cousin is a consumer, and the post office is a message broker.

Figure 1: Analogy for message broker

RabbitMQ as message broker

Now we know that the purpose of a message broker is to route messages from a producer to a consumer. Let’s examine one such message broker — RabbitMQ. It’s one of the most extensively used message brokers these days.

The way RabbitMQ routes messages depends upon the messaging protocol it implements. RabbitMQ supports multiple messaging protocols. However, the one we are interested in is AMQP. It is an acronym for Advanced Message Queuing Protocol.

So without any further ado let’s have a closer look at the AMQP protocol model.

Advanced Message Queuing Protocol

The conceptual model of AMQP is quite simple and straightforward. It has three entities:

1. Queue
2. Binding
3. Exchange

When a publisher pushes a message to RabbitMQ, it first arrives at an exchange. The exchange then distributes copies of these messages to variously connected queues. Finally, consumers receive these messages.

Consider a message as a piece of data. It is necessarily a package with a payload and some meta-data. The payload contains full data whereas meta-data are properties used by RabbitMQ.

Figure 2 depicts a graphical representation of the AMQP model.

Figure 2: AMQP model

AMQP is a programmable protocol. Programmers have the liberty to use libraries to configure entities (exchange, binding, and queue) as per their own needs. A RabbitMQ admin has no role in setting up these entities.

There are plenty of libraries available to work with RabbitMQ. You can choose from Nodejs, Python, .Net, Java, and many more.

Queues

These queues are somehow similar to the queues from our data structure classes. RabbitMQ queues also follow FIFO — First-In-First-Out methodology. A queue is a place where RabbitMQ stores messages/data.

Programmers can configure queues through available programming libraries. You can make a queue durable ( with the Durability property) to safeguard your data in case the broker crashes. You can also provide a name(with the Name property) to a queue. Other than Name and Durability, a queue has a few other properties like auto-delete, exclusive, and arguments.

Before moving any further, it’s important to understand who is a direct consumer of these queues. Moreover, how many ways can a user consume messages from a queue?

Consumers

Consumers are the ones who are going to use messages stored in a queue. It is possible to connect more than one consumer to a queue at a time. Consumers can either pull the message from the queue by pooling it or queues can even push the message to various connect consumers.

Bindings

Bindings are the rules that a queue defines while establishing a connection with an exchange. You can have a queue connected to multiple exchanges. Every queue is also connected to a default exchange. An exchange will use these bindings to route messages to queues.

Exchanges and their types

An Exchange is a gateway to RabbitMQ for your messages. The distance the message has to travel inside RabbitMQ depends on the type of exchange. Primarily there are four types.

• Direct
• Fanout
• Topic
• Header

Direct

The name explains it all! — A direct exchange delivers a message directly to the queues that satisfy the below condition:

Routing key == Binding key

A routing key is an attribute of the message. On the other hand, a binding key is something you specify while creating a binding between a queue and an exchange.

Figure 3 is a visual explanation of how messages flow while using a direct exchange.

A message originates from a producer (green circle) with a routing key — img.resize. Once it reaches the exchange (Orange circle), the exchange will try to find all queues with binding key — img.resize. In case of a match, the message is pushed to all matched queues (resize in our case). If there is no match found, the message can be sent back to the producer or can even be discarded. We are lucky that we found a match in our example ?

[gif image]

Once the message reaches the desired queue (resize in our case), they are distributed in round-robin fashion to all the connected consumers (resizer.1/resizer.2 in our case).

By distributing messages in a round-robin fashion, RabbitMQ makes sure that the messages are load balanced.

You must have noticed that the queue named crop is not receiving any messages. Because the routing key in this example is img.resize. To send messages to this queue, we need to send messages with a routing key that would match the binding key (say img.crop for instance).

Fanout

A Fanout exchange ignores routing keys and distributes a message to all the connected queues. No wonder it is called Fanout (blowing messages to all connected queues! ?).

One of the use cases for this type of exchange is message broadcast.

Please note that RabbitMQ will still do round robin if there is more than one consumer of the queue.

Topic

A topic exchange routes a message by matching routing key with a pattern in the binding key.

Routing key == Pattern in binding key.

RabbitMQ uses two wild card characters for pattern matching * and #. Use a * to match 1 word and a # to match 0 or more words.

Figure 5 is a visual depiction of a topic exchange. Messages with routing key — logs.error will match patterns — logs.error and logs.*. Hence these messages will end up in the queues — only error and all logs.

Whereas for the producer at the bottom-left, messages with routing key— logs.success will match patterns of binding key #success and logs.* . Hence these messages will end up in the queues — all logsand only success .

[gif]

This type of exchange has a vast range of use cases. It can be used in the publish-subscribe pattern, distributing relevant data to desiring workers processes and many more.

Header

A header is a particular type of exchange that routes messages based on keys present in the message header. It overlooks routing key attribute of the message.

When creating bindings for a header exchange, it is possible to bind a queue to match more than one header. In such a case, RabbitMQ should know from the producer if it should match all or any of these keys.

A producer/application can do this by providing an extra flag called ‘x-match’. ‘x-match’ can have any or all values. The first one mandates that only one value should match while the latter mandates that all must match.

Message Acknowledgement

Once a message reaches its destination, the broker should delete the message from the queue. It is necessary because a queue overflow can occur if it keeps accumulating messages.

Before deleting any message, the broker must have a delivery acknowledgment. There are two possible ways to acknowledge message delivery.

1. Automatic acknowledgment: Once a consumer receives the message
2. Explicit acknowledgment: When a consumer sends back an acknowledgment

In most cases, explicit acknowledgment is used as it makes sure that the consumer has consumed the message without any failover.

What’s next

RabbitMQ is a very mature and useful product. This article is only a high-level introduction to RabbitMQ. I simplified the concepts to provide a reference point for you to move further. Visit the RabbitMQ website for more complex topics.

Hope you like the article. Don’t forget to clap(or applaud ?). Follow to read my upcoming stories. Until next time, keep Queuing.

Black Friday’s right around the corner, and the new Microsoft Surface Studio out in stores (I’m a loyal windows guy ?). So let’s talk about everyone’s favorite shopping past-time: waiting in line. And that age-old data structure, the queue.

Feel free to share this post with your friends who’ll be heading out to get the latest and greatest. But be warned — people have been known to forget how queues work on Black Friday.

Queues

A queue is a line (yep, the same one from kindergarten…no cutting still!)

Additions (enqueues) always add to the back of the line

Removals (dequeues) always remove from the front of the line

Queues follow the pattern of First item In is the First item Out (FIFO).

Example use cases

• Resolving simultaneous server requests from multiple users, such as 3 people buying the last ticket for a plane at almost the same time
• Queuing up data during a breadth-first search.

Let’s tackle the first use case by helping Microsoft create a queue data structure to manage all their requests for the new Surface Studio. I’m too busy coding and writing these posts to go get one myself, so if you’re a Microsoft representative reading this, please feel free to ship one my way. ?

Before we get started, a quick note on JavaScript arrays. Similar to stacks, JavaScript arrays naturally have the functionality of a queue built in.

How to represent queues using JavaScript arrays

Enqueue adds to the back of the array:

Array.push(someVal)

Dequeue removes and returns first item in array:

Array.shift()

If for some reason you’re feeling rebellious (what coder doesn’t ?) you could add to the front of the array, then remove from the back.

Enqueue adds item to the front of the array:

Array.unshift(someVal)

Dequeue removes item from the back of the array:

Array.pop()

That said, for the sake of being thorough, you’re going to rebuild it using a JavaScript Object.

So first thing you need to do for Microsoft is to create the actual Queue where you’re going to hold the individual members who click the buy button on their website.

class Queue{  constructor(){    this._storage = {};    this._start = -1; //replicating 0 index used for arrays    this._end = -1; //replicating 0 index used for arrays  }    size(){   return this._end - this._start;  }}

let appleQueue = new Queue();

As a quick reminder the _ just means this is a private variable, and shouldn’t be accessed directly.

Unlike the stack data structure, where additions and removals happen on the same side, the nature of the queue requires us to keep track of both ends. Because of that, you create the start variable to always track the front of the queue, and the end variable to track the end of the queue.

Lastly, the easiest way to keep track of a queue’s size (without creating an unnecessary counter variable) is to keep track of the difference between your start and end points.

First, you should create a way for people who click buy to be added to the queue. You can do this via the enqueue method:

class Queue{  constructor(){    this._storage = {};    this._start = -1; //replicating 0 index used for arrays    this._end = -1; //replicating 0 index used for arrays  }    enqueue(val){    this._storage[++this._end] = val;          //++this._end just means increment the end variable first    //It's equivalent to    //this._end++   //->    //this._storage[this._end] = val;  }    size(){   return this._end - this._start;  }}

let microsoftQueue = new Queue();

microsoftQueue.enqueue("{user: ILoveWindows@gmail.com}")microsoftQueue.enqueue("{user: cortanaIsMyBestFriend@hotmail.com}")microsoftQueue.enqueue("{user: InternetExplorer8Fan@outlook.com}")microsoftQueue.enqueue("{user: IThrowApplesOutMyWindow@yahoo.com}")

Great! Now your microsoftQueue storage is going to look a little something like this:

{

  0: "{email: ILoveWindows@gmail.com}"

  1: "{email: cortanaIsMyBestFriend@hotmail.com}"

  2: "{email: InternetExplorer8Fan@outlook.com}"

  3: "{email: IThrowApplesOutMyWindow@yahoo.com}"

}

So a quick note on the way users are being represented above ({user: …}).

When a user clicks the buy button on the client side, they’re sending all their relevant information to the server, which will handle the request. When data is often exchanged between systems, such as the client and server side, it’s most commonly sent as JSON (JavaScript Object Notation), via Ajax.

This is similar to JavaScript objects, in that it’s just a stringified version of key-value pairs. For those not familiar with JavaScript, it’s similar to a dictionary or hash table (which we’ll get to later in this series). For more information about this , there’s a great post here on StackOverflow by Andreas Grech.

Now back to your queue.

Thanks to the queue you created, Microsoft now has an efficient way of tracking all of the people who have purchased the Surface Studio, and in chronological order in which they purchased it. To make sure these people are served in the correct order, you need to create an accurate dequeue method that keeps track of the order of the buyers, and removes them from the queue once they’ve been served.

class Queue{  constructor(){    this._storage = {};    this._start = -1; //replicating 0 index used for arrays    this._end = -1; //replicating 0 index used for arrays  }    enqueue(val){    this._storage[++this._end] = val;   }

  dequeue(){    if(this._end > this._start){ //check if there are values      let nextUp = this._storage[++this._start];      delete this._storage[this._start];      return nextUp;    }  }      size(){   return this._end - this._start;  }}

let microsoftQueue = new Queue();

microsoftQueue.enqueue("{user: ILoveWindows@gmail.com}")microsoftQueue.enqueue("{user: cortanaIsMyBestFriend@hotmail.com}")microsoftQueue.enqueue("{user: InternetExplorer8Fan@outlook.com}")microsoftQueue.enqueue("{user: IThrowApplesOutMyWindow@yahoo.com}")

//Function to send everyone their Surface Studio!let sendSurface = recepient => {   sendTo(recepient);}

//When your server is ready to handle this queue, execute this:

while(microsoftQueue.size() > 0){  sendSurface(microsoftQueue.dequeue());}

And there it is folks! Everyone who was waiting in the microsoftQueue now gets their awesome new Surface Studio thanks to you.

To be thorough, there are definitely some quick optimizations that can make code work more logically.

1. You can reset your start and end values to 0 once everyone in the queue has been served. It’s unlikely that your queue would ever hit the “max” JavaScript number, but it’s better to be safe than sorry.
2. You can switch out the “ end > start check ” with the size method, thanks to 0 being evaluated as “false” due to JavaScript type coercion. Read all about it here.

dequeue(){    if(this.size()){ //0 is a falsey value...coerced to return false      let nextUp = this._storage[++this._start];      delete this._storage[this._start];

      if(!this.size()){ //Recheck after incrementing (!0 == true)        this._start = -1;        this._end = -1;       }            return nextUp;    }}

And there you go, you’ve finished writing your basic Queue!

A time complexity analysis on the queue methods

Here’s the code again:

class Queue{  constructor(){    this._storage = {};    this._start = -1; //replicating 0 index used for arrays    this._end = -1; //replicating 0 index used for arrays  }    enqueue(val){    this._storage[++this._end] = val;   }

  dequeue(){    if(this.size()){ /      let nextUp = this._storage[++this._start];      delete this._storage[this._start];

      if(!this.size()){         this._start = -1;        this._end = -1;       }            return nextUp;    }  }    size(){   return this._end - this._start;  }}

The same logic for stacks also applies here:

Enqueue (addition) is O(1). Since you’ll always know where the end of the queue is (thanks to your end variable), you don’t have to iterate to add an item.

Dequeue (removal) is O(1). No iteration is necessary for removal since you always have the current start position.

Size is O(1). The size is always known thanks to your start and end variables.

One really important thing to note here is that queues aren’t meant to be infinite, although our queue class and JavaScript array will allow you to keep on adding items until the system runs out of memory.

One way to optimize is by making a space-limited array to create a circular queue. Damian Gordon provides a really good video walk-through on YouTube. This will also be handy for when we get to hash tables in future articles!

Time for a quick recap

Queues:

1. Follow a First In First Out (FIFO) pattern
2. Have a start and end property to track the front and back of your queue
3. Have an enqueue (add) and dequeue (remove) method to manage the contents of your queue
4. Have a size property that allows you to track how large your queue is

Here’s a quick challenge

Using what you now know about Stacks and what you learned today about Queues, try re-implementing a Queue using just stacks.

As a quick hint, you’ll only need two stacks.

Thanks to Jason Holtkamp for coming up with this quick challenge!

Further reading

Wikipedia as always ?

This Wikipedia article on priority queue. We’ll come back to this in future articles.

A nice demo by Larry Serflaten on Khan Academy, where he uses push and pull in place of enqueue and dequeue.

And here’s the Answer for the quick challenge. Only look at this after trying it out for a bit yourself. You can also check out Levent Divilioglu’s answer for a fantastic graphical representation.