Message Queue Patterns for Distributed Systems

Message queues decouple services and enable asynchronous communication. Here are the essential patterns for building reliable distributed systems.

Message Queue Concepts#

Queue: FIFO storage for messages
Exchange: Routes messages to queues
Binding: Rules connecting exchanges to queues
Consumer: Processes messages from queue
Producer: Sends messages to exchange

Patterns:
- Work Queue: Distribute tasks among workers
- Pub/Sub: Broadcast to all subscribers
- Routing: Send to specific queues
- Topics: Pattern-based routing
- RPC: Request/reply over queue

Work Queue Pattern#

import { Queue, Worker } from 'bullmq';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);

// Producer - Add jobs to queue
const emailQueue = new Queue('emails', { connection: redis });

async function queueEmail(to: string, subject: string, body: string): Promise<string> {
  const job = await emailQueue.add(
    'send-email',
    { to, subject, body },
    {
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 1000,
      },
      removeOnComplete: 100,
      removeOnFail: 1000,
    }
  );

  return job.id!;
}

// Consumer - Process jobs
const emailWorker = new Worker(
  'emails',
  async (job) => {
    const { to, subject, body } = job.data;

    console.log(`Processing email to ${to}`);

    await sendEmail(to, subject, body);

    return { sent: true, timestamp: new Date() };
  },
  {
    connection: redis,
    concurrency: 5,
    limiter: {
      max: 100,
      duration: 60000, // 100 per minute
    },
  }
);

emailWorker.on('completed', (job, result) => {
  console.log(`Email ${job.id} sent successfully`);
});

emailWorker.on('failed', (job, error) => {
  console.error(`Email ${job?.id} failed:`, error.message);
});

Pub/Sub Pattern#

import Redis from 'ioredis';

// Publisher
class EventPublisher {
  constructor(private redis: Redis) {}

  async publish(channel: string, event: any): Promise<void> {
    const message = JSON.stringify({
      id: crypto.randomUUID(),
      timestamp: new Date().toISOString(),
      data: event,
    });

    await this.redis.publish(channel, message);
  }
}

// Subscriber
class EventSubscriber {
  private subscriber: Redis;
  private handlers: Map<string, ((event: any) => Promise<void>)[]> = new Map();

  constructor(redisUrl: string) {
    this.subscriber = new Redis(redisUrl);

    this.subscriber.on('message', async (channel, message) => {
      const event = JSON.parse(message);
      const handlers = this.handlers.get(channel) || [];

      for (const handler of handlers) {
        try {
          await handler(event);
        } catch (error) {
          console.error(`Handler error for ${channel}:`, error);
        }
      }
    });
  }

  subscribe(channel: string, handler: (event: any) => Promise<void>): void {
    if (!this.handlers.has(channel)) {
      this.handlers.set(channel, []);
      this.subscriber.subscribe(channel);
    }
    this.handlers.get(channel)!.push(handler);
  }
}

// Usage
const publisher = new EventPublisher(redis);
const subscriber = new EventSubscriber(process.env.REDIS_URL!);

subscriber.subscribe('orders', async (event) => {
  console.log('Order event:', event.data);
  await processOrder(event.data);
});

subscriber.subscribe('orders', async (event) => {
  await sendNotification(event.data);
});

await publisher.publish('orders', {
  type: 'ORDER_CREATED',
  orderId: '123',
  customerId: 'cust_456',
});

Dead Letter Queue#

// Main queue with dead letter handling
const orderQueue = new Queue('orders', {
  connection: redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,
    },
  },
});

// Dead letter queue
const deadLetterQueue = new Queue('orders-dlq', { connection: redis });

const orderWorker = new Worker(
  'orders',
  async (job) => {
    const order = job.data;

    try {
      await processOrder(order);
    } catch (error) {
      // Check if should go to DLQ
      if (job.attemptsMade >= job.opts.attempts!) {
        await deadLetterQueue.add('failed-order', {
          originalJob: job.data,
          error: error.message,
          failedAt: new Date(),
          attempts: job.attemptsMade,
        });
      }
      throw error;
    }
  },
  { connection: redis }
);

// DLQ processor - manual review or retry
const dlqWorker = new Worker(
  'orders-dlq',
  async (job) => {
    const { originalJob, error } = job.data;

    // Log for investigation
    console.error('DLQ item:', {
      order: originalJob,
      error,
    });

    // Optionally notify support
    await notifySupport(job.data);
  },
  { connection: redis }
);

// Retry from DLQ
async function retryDeadLetter(jobId: string): Promise<void> {
  const job = await deadLetterQueue.getJob(jobId);
  if (!job) throw new Error('Job not found');

  // Re-queue original job
  await orderQueue.add('process-order', job.data.originalJob);

  // Remove from DLQ
  await job.remove();
}

Priority Queues#

// Add jobs with priority
await orderQueue.add(
  'process-order',
  { orderId: '123', type: 'express' },
  { priority: 1 } // Lower number = higher priority
);

await orderQueue.add(
  'process-order',
  { orderId: '456', type: 'standard' },
  { priority: 5 }
);

// Separate queues for different priorities
const highPriorityQueue = new Queue('orders-high', { connection: redis });
const normalPriorityQueue = new Queue('orders-normal', { connection: redis });

// Process high priority first
const highWorker = new Worker(
  'orders-high',
  processOrder,
  { connection: redis, concurrency: 10 }
);

const normalWorker = new Worker(
  'orders-normal',
  processOrder,
  { connection: redis, concurrency: 5 }
);

Delayed Messages#

// Schedule job for later
await emailQueue.add(
  'send-reminder',
  { userId: '123', type: 'cart-abandonment' },
  {
    delay: 24 * 60 * 60 * 1000, // 24 hours
  }
);

// Recurring jobs
await emailQueue.add(
  'daily-digest',
  { type: 'digest' },
  {
    repeat: {
      pattern: '0 9 * * *', // Daily at 9 AM
    },
  }
);

// Job that runs once at specific time
const futureDate = new Date('2024-12-25T10:00:00Z');
await emailQueue.add(
  'christmas-email',
  { campaign: 'xmas-2024' },
  {
    delay: futureDate.getTime() - Date.now(),
  }
);

Request/Reply Pattern#

// RPC over message queue
class RpcClient {
  private responsePromises: Map<string, {
    resolve: (value: any) => void;
    reject: (error: Error) => void;
  }> = new Map();

  constructor(
    private requestQueue: Queue,
    private responseQueue: Queue
  ) {
    // Listen for responses
    new Worker(
      'rpc-responses',
      async (job) => {
        const { correlationId, result, error } = job.data;
        const pending = this.responsePromises.get(correlationId);

        if (pending) {
          if (error) {
            pending.reject(new Error(error));
          } else {
            pending.resolve(result);
          }
          this.responsePromises.delete(correlationId);
        }
      },
      { connection: redis }
    );
  }

  async call<T>(method: string, params: any, timeout = 30000): Promise<T> {
    const correlationId = crypto.randomUUID();

    const promise = new Promise<T>((resolve, reject) => {
      this.responsePromises.set(correlationId, { resolve, reject });

      // Timeout
      setTimeout(() => {
        if (this.responsePromises.has(correlationId)) {
          this.responsePromises.delete(correlationId);
          reject(new Error('RPC timeout'));
        }
      }, timeout);
    });

    await this.requestQueue.add('rpc-request', {
      correlationId,
      method,
      params,
      replyTo: 'rpc-responses',
    });

    return promise;
  }
}

// RPC Server
const rpcServer = new Worker(
  'rpc-requests',
  async (job) => {
    const { correlationId, method, params, replyTo } = job.data;
    const replyQueue = new Queue(replyTo, { connection: redis });

    try {
      const result = await executeMethod(method, params);
      await replyQueue.add('response', { correlationId, result });
    } catch (error) {
      await replyQueue.add('response', {
        correlationId,
        error: error.message,
      });
    }
  },
  { connection: redis }
);

// Usage
const client = new RpcClient(requestQueue, responseQueue);
const result = await client.call('calculatePrice', { productId: '123' });

Idempotency#

// Ensure messages are processed exactly once
class IdempotentWorker {
  constructor(
    private redis: Redis,
    private ttl = 24 * 60 * 60 // 24 hours
  ) {}

  async processOnce<T>(
    messageId: string,
    processor: () => Promise<T>
  ): Promise<T | null> {
    const key = `processed:${messageId}`;

    // Check if already processed
    const existing = await this.redis.get(key);
    if (existing) {
      console.log(`Message ${messageId} already processed`);
      return JSON.parse(existing);
    }

    // Process
    const result = await processor();

    // Mark as processed
    await this.redis.setex(key, this.ttl, JSON.stringify(result));

    return result;
  }
}

const idempotentWorker = new IdempotentWorker(redis);

const worker = new Worker(
  'orders',
  async (job) => {
    return idempotentWorker.processOnce(job.id!, async () => {
      return processOrder(job.data);
    });
  },
  { connection: redis }
);

Monitoring#

// Queue metrics
async function getQueueMetrics(queue: Queue) {
  const [waiting, active, completed, failed, delayed] = await Promise.all([
    queue.getWaitingCount(),
    queue.getActiveCount(),
    queue.getCompletedCount(),
    queue.getFailedCount(),
    queue.getDelayedCount(),
  ]);

  return { waiting, active, completed, failed, delayed };
}

// Expose metrics endpoint
app.get('/metrics/queues', async (req, res) => {
  const metrics = {
    orders: await getQueueMetrics(orderQueue),
    emails: await getQueueMetrics(emailQueue),
  };

  res.json(metrics);
});

Best Practices#

Reliability:
✓ Use acknowledgments
✓ Implement dead letter queues
✓ Make handlers idempotent
✓ Set appropriate retries

Performance:
✓ Batch when possible
✓ Tune concurrency
✓ Monitor queue depth
✓ Scale consumers dynamically

Operations:
✓ Log job processing
✓ Alert on queue buildup
✓ Track processing time
✓ Plan for failures

Message queues enable reliable, scalable distributed systems. Use work queues for task distribution, pub/sub for broadcasts, and dead letter queues for failure handling. Always ensure idempotency and monitor queue health.