Building a Custom Message Broker in Node.js: Implementing Pub/Sub and Consumer Groups from Scratch

Modern distributed systems rely heavily on message brokers like RabbitMQ, Apache Kafka, or AWS SQS to decouple services, balance loads, and ensure reliable asynchronous communication. But how do these systems actually work under the hood?

Instead of pulling in a massive infrastructure dependency for a simple project, or blindly trusting a black-box library, building your own lightweight message broker is one of the best ways to master distributed systems engineering. In this post, we will design and implement a custom message broker in Node.js from scratch using Redis Streams as our underlying storage engine.

By the end of this guide, you will understand how to build:

• A Publish/Subscribe (Pub/Sub) engine for broadcast messaging.
• Consumer Groups for load balancing message processing across multiple worker instances.
• Manual Acknowledgments (ACKs) and pending entry lists (PEL) to guarantee at-least-once delivery semantics.

1. Architectural Overview

To build a robust message broker, we need to move beyond volatile memory structures and transient pub/sub channels (like traditional Redis Pub/Sub, which drops messages if no subscribers are listening). We need persistence, ordering guarantees, and offset tracking.

Redis Streams (XADD, XREAD, XREADGROUP) give us precisely these primitives. They act like append-only logs:

+-------------------------------------------------------------+
|                        Redis Stream                         |
|  [ID: 16900-0] -> [ID: 16900-1] -> [ID: 16900-2] (Log Tail)  |
+-------------------------------------------------------------+
         |                                           ^
         v (Consume)                                 | (ACK)
+-------------------+                      +-------------------+
|  Consumer Group   |                      |  Pending Entries  |
|  [Worker A, B, C] |                      |  (Awaiting ACK)   |
+-------------------+                      +-------------------+

Our Node.js wrapper will abstract these Redis primitives into a clean, object-oriented API consisting of a MessageBroker class, a Publisher, and a ConsumerWorker with consumer group semantics.

2. Setting Up the Project

First, initialize a new Node.js project and install ioredis, a robust, feature-rich Redis client for Node.js.

mkdir custom-message-broker
cd custom-message-broker
npm init -y
npm install ioredis dotenv

Create a file structure like this:

custom-message-broker/
├── broker.js
├── publisher.js
├── worker.js
└── package.json

3. Building the Core Broker Class

The MessageBroker class will manage our connection to Redis and expose methods for publishing messages, creating consumer groups, reading from streams, and acknowledging processed messages.

Create broker.js:

const Redis = require('ioredis');

class MessageBroker {
  constructor(redisConfig = {}) {
    this.redis = new Redis(redisConfig);
    this.subscriberRedis = new Redis(redisConfig); // Dedicated connection for blocking reads
  }

  /**
   * Publish a message to a specific stream/topic
   * @param {string} streamName 
   * @param {Object} payload 
   */
  async publish(streamName, payload) {
    // Serialize payload values to strings as required by Redis streams
    const flattenedPayload = Object.entries(payload).flatMap(([key, value]) => [
      key,
      typeof value === 'object' ? JSON.stringify(value) : String(value)
    ]);

    const messageId = await this.redis.xadd(streamName, '*', ...flattenedPayload);
    return messageId;
  }

  /**
   * Create a consumer group for a stream
   * @param {string} streamName 
   * @param {string} groupName 
   */
  async createConsumerGroup(streamName, groupName) {
    try {
      // '0' means read from the very beginning of the stream, '$' means only new messages
      await this.redis.xgroup('CREATE', streamName, groupName, '0', 'MKSTREAM');
      console.log(`[Broker] Consumer group '${groupName}' created for stream '${streamName}'.`);
    } catch (err) {
      if (err.message.includes('BUSYGROUP')) {
        console.log(`[Broker] Consumer group '${groupName}' already exists.`);
      } else {
        throw err;
      }
    }
  }

  /**
   * Consume messages as part of a consumer group
   * @param {string} streamName 
   * @param {string} groupName 
   * @param {string} consumerName 
   * @param {number} count 
   * @param {number} blockMs 
   */
  async consumeGroup(streamName, groupName, consumerName, count = 1, blockMs = 5000) {
    try {
      // XREADGROUP GROUP groupName consumerName COUNT count BLOCK blockMs STREAMS streamName >
      // '>' means fetch only messages never delivered to other consumers in this group
      const response = await this.subscriberRedis.xreadgroup(
        'GROUP',
        groupName,
        consumerName,
        'COUNT',
        count,
        'BLOCK',
        blockMs,
        'STREAMS',
        streamName,
        '>'
      );

      if (!response) return [];

      // Parse Redis stream response format into clean JS objects
      const [_, rawMessages] = response[0];
      return rawMessages.map(([id, fields]) => {
        const data = {};
        for (let i = 0; i < fields.length; i += 2) {
          try {
            data[fields[i]] = JSON.parse(fields[i + 1]);
          } catch {
            data[fields[i]] = fields[i + 1];
          }
        }
        return { id, data };
      });
    } catch (err) {
      console.error('[Broker] Error reading from consumer group:', err);
      return [];
    }
  }

  /**
   * Acknowledge that a message has been successfully processed
   * @param {string} streamName 
   * @param {string} groupName 
   * @param {string} messageId 
   */
  async acknowledge(streamName, groupName, messageId) {
    await this.redis.xack(streamName, groupName, messageId);
  }

  async disconnect() {
    await this.redis.quit();
    await this.subscriberRedis.quit();
  }
}

module.exports = MessageBroker;

Key Design Choices in MessageBroker:

1. Dual Redis Connections: Redis blocks client connections during operations like XREADGROUP with a BLOCK argument. If we used the same connection for publishing and consuming, our application would lock up. Having a dedicated subscriberRedis instance prevents deadlocks.
2. The > Specifier: When querying via XREADGROUP, passing > tells Redis: “Give me only messages that have not yet been assigned to any other consumer in this group.”
3. Payload Serialization: Redis stream entries are flat key-value pairs of strings. Our wrapper transparently handles object serialization and deserialization so developers can pass rich JSON payloads.

4. Implementing the Publisher

Now, let’s create a simple publisher script that pushes jobs/messages into our broker at regular intervals.

Create publisher.js:

const MessageBroker = require('./broker');

async function runPublisher() {
  const broker = new MessageBroker();
  const streamName = 'job-stream';

  console.log('[Publisher] Starting publisher...');

  let counter = 1;
  setInterval(async () => {
    const payload = {
      jobId: counter,
      task: 'generate-report',
      timestamp: new Date().toISOString(),
      metadata: { user: `user_${counter}`, priority: counter % 2 === 0 ? 'high' : 'low' }
    };

    const messageId = await broker.publish(streamName, payload);
    console.log(`[Publisher] Sent message ID: ${messageId} | Job ID: ${counter}`);
    counter++;
  }, 2000);
}

runPublisher().catch(console.error);

5. Implementing Consumer Groups and Workers

Consumer groups allow multiple instances of a worker service to scale horizontally. Redis automatically load-balances messages across all active consumers in the same group.

Furthermore, Redis maintains a Pending Entries List (PEL) for each consumer group. If a worker crashes after reading a message but before acknowledging it (XACK), that message remains in the PEL and can eventually be reclaimed or retried.

Create worker.js:

const MessageBroker = require('./broker');

async function runWorker() {
  const broker = new MessageBroker();
  const streamName = 'job-stream';
  const groupName = 'analytics-cluster';
  
  // Accept consumer identifier via CLI argument (e.g., node worker.js worker-1)
  const consumerName = process.argv[2] || `worker-${process.pid}`;

  // Ensure consumer group exists before attempting to read
  await broker.createConsumerGroup(streamName, groupName);

  console.log(`[Worker: ${consumerName}] Ready and listening for messages...`);

  let isRunning = true;
  process.on('SIGINT', async () => {
    console.log(`[Worker: ${consumerName}] Shutting down gracefully...`);
    isRunning = false;
    await broker.disconnect();
    process.exit(0);
  });

  while (isRunning) {
    try {
      // Block for up to 5 seconds waiting for new messages
      const messages = await broker.consumeGroup(streamName, groupName, consumerName, 1, 5000);

      for (const message of messages) {
        console.log(`[Worker: ${consumerName}] Processing message ID: ${message.id}`, message.data);

        // Simulate processing delay (e.g., calling an external API or heavy compute)
        await simulateWork(1500);

        // Acknowledge successful processing
        await broker.acknowledge(streamName, groupName, message.id);
        console.log(`[Worker: ${consumerName}] Acknowledged message ID: ${message.id}`);
      }
    } catch (err) {
      console.error(`[Worker: ${consumerName}] Error in processing loop:`, err);
    }
  }
}

function simulateWork(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

runWorker().catch(console.error);

6. Testing the Message Broker

To see consumer groups and load balancing in action, we can run multiple worker terminals alongside our publisher.

Step 1: Start Redis

Ensure your local Redis instance is running:

redis-server

Step 2: Start the Publisher

node publisher.js

Step 3: Start Worker 1

In a separate terminal, launch the first worker:

node worker.js worker-alpha

Step 4: Start Worker 2

In yet another terminal, launch a second worker under the same consumer group (analytics-cluster):

node worker.js worker-beta

Observation:

Look at your worker terminals. You will notice that worker-alpha and worker-beta alternately process incoming jobs from the publisher. Redis Streams distributes messages round-robin style across all consumers registered within the analytics-cluster group. If you kill worker-alpha, worker-beta seamlessly picks up the entire message stream.

7. Advanced Considerations & Production Readiness

While our custom broker is fully functional for development and educational purposes, scaling this pattern to production requires handling edge cases that off-the-shelf brokers handle out-of-the-box:

1. Dead Letter Queues (DLQ): If a message fails processing repeatedly (e.g., due to malformed data), it shouldn’t block the PEL forever. You should track delivery retry counts using XPENDING and move poisoned messages to a dedicated stream:dlq after $N$ failures.
2. Pending Entry Reclamation (XCLAIM): If a worker crashes permanently while holding unacknowledged messages in the PEL, those messages will sit there indefinitely. A background cron job should periodically check older pending messages using XPENDING and claim them for recovery using XCLAIM.
3. Stream Trimming (XADD MAXLEN): Redis streams grow indefinitely unless trimmed. In production, ensure you cap your streams using approximate trimming (XADD mystream MAXLEN ~ 10000 * ...) to prevent Redis from running out of memory.

Conclusion

By leveraging Redis Streams and Node.js, we’ve built a lightweight, reliable message broker from scratch complete with publish-subscribe semantics, consumer groups, load balancing, and manual acknowledgements. Understanding these primitives gives you profound insight into how enterprise-grade distributed messaging systems work under the hood, empowering you to design more resilient backend architectures.