Queue & Pub/Sub in JavaScript

In this article we will know how to use queues and pubsubs in javascript. A message queue is a communication method used in software systems where one program (the producer) sends messages to a queue, and another program (the consumer) retrieves and processes them later.

How It Works

• Producer sends a message to the queue.

• The message is stored temporarily.

• Consumer reads and processes the message.

• The message is removed from the queue after processing.

Why Use a Message Queue?

Decoupling

Systems don’t need to know about each other directly.

Scalability

You can add more consumers to process more messages.

Reliability

Messages can be saved until successfully processed.

Asynchronous Processing

Tasks like sending emails or processing payments can happen in the background.

Implementation of Queues in node js

Docker YAML For RabbitMQ

version: "3.8"

services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    restart: always
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin123
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
    networks:
      - backend

volumes:
  rabbitmq_data:

networks:
  backend:

Run the YAML file by the following command.

docker compose up -d

Design Pub/Sub With RabbitMQ in Node Js

Suppose there are 2 services in our backend - Auth and email. So we are bulding pub-sub for after register we send a event to email service to send otp email.

npm install amqplib express uuid

auth-service/rabbit.js

const amqp = require("amqplib");

let channel;

async function connectRabbit() {
  const connection = await amqp.connect(
    "amqp://admin:admin123@localhost:5672"
  );

  channel = await connection.createChannel();

  await channel.assertExchange("user_registered_exchange", "fanout", {
    durable: true
  });

  console.log("Auth service connected to RabbitMQ");
}

function publishUserRegistered(user) {
  const message = {
    event: "USER_REGISTERED",
    data: user,
    createdAt: new Date()
  };

  channel.publish(
    "user_registered_exchange",
    "",
    Buffer.from(JSON.stringify(message)),
    { persistent: true }
  );

  console.log("USER_REGISTERED event published");
}

module.exports = { connectRabbit, publishUserRegistered };

email-service/rabbit.js

const amqp = require("amqplib");
const nodemailer = require("nodemailer");

const EXCHANGE = "user_exchange";
const QUEUE = "email_queue";
const ROUTING_KEY = "user.registered.email";

async function startEmailService() {
  const connection = await amqp.connect(
    "amqp://admin:admin123@localhost:5672"
  );

  const channel = await connection.createChannel();

  // 1️⃣ Assert exchange
  await channel.assertExchange(EXCHANGE, "direct", {
    durable: true
  });

  // 2️⃣ Create queue
  await channel.assertQueue(QUEUE, {
    durable: true
  });

  // 3️⃣ Bind queue with routing key
  await channel.bindQueue(QUEUE, EXCHANGE, ROUTING_KEY);

  channel.prefetch(1);

  console.log("Email service waiting for routed messages...");

  channel.consume(QUEUE, async (msg) => {
    if (!msg) return;

    const content = JSON.parse(msg.content.toString());
    console.log("Received:", content);

    try {
      if (content.event === "USER_REGISTERED") {
        await sendWelcomeEmail(content.data);
      }

      channel.ack(msg); //acknowledge that message has been processed
    } catch (err) {
      console.error("Email failed:", err);
      channel.nack(msg, false, true);
    }
  });
}

async function sendWelcomeEmail(user) {
  console.log(`Sending email to ${user.email}`);

  const transporter = nodemailer.createTransport({
    service: "gmail",
    auth: {
      user: "your@gmail.com",
      pass: "app-password"
    }
  });

  await transporter.sendMail({
    from: "your@gmail.com",
    to: user.email,
    subject: "Welcome!",
    text: `Hello ${user.name}, welcome!`
  });

  console.log("Email sent");
}

startEmailService();