Message Queues: RabbitMQ and AWS SQS Patterns

Message queues enable asynchronous, decoupled communication. This guide covers implementing reliable messaging.

RabbitMQ Basics#

Publisher#

import amqp from 'amqplib';

const connection = await amqp.connect(process.env.RABBITMQ_URL);
const channel = await connection.createChannel();

// Ensure queue exists
await channel.assertQueue('orders', { durable: true });

// Publish message
function publishOrder(order: Order) {
  channel.sendToQueue(
    'orders',
    Buffer.from(JSON.stringify(order)),
    {
      persistent: true,
      contentType: 'application/json',
      messageId: order.id,
    }
  );
}

Consumer#

await channel.assertQueue('orders', { durable: true });
await channel.prefetch(10); // Process 10 at a time

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

  try {
    const order = JSON.parse(msg.content.toString());
    await processOrder(order);
    channel.ack(msg);
  } catch (error) {
    // Requeue on failure (or send to DLQ)
    channel.nack(msg, false, true);
  }
});

Exchange Patterns#

// Topic exchange for routing
await channel.assertExchange('events', 'topic', { durable: true });

// Publish with routing key
channel.publish('events', 'order.created', Buffer.from(JSON.stringify(event)));
channel.publish('events', 'order.shipped', Buffer.from(JSON.stringify(event)));

// Subscribe to patterns
await channel.bindQueue('notifications', 'events', 'order.*');
await channel.bindQueue('analytics', 'events', '#'); // All events

AWS SQS#

Standard Queue#

import { SQS } from '@aws-sdk/client-sqs';

const sqs = new SQS({ region: 'us-east-1' });

// Send message
await sqs.sendMessage({
  QueueUrl: process.env.QUEUE_URL,
  MessageBody: JSON.stringify(payload),
  MessageAttributes: {
    EventType: { DataType: 'String', StringValue: 'order.created' },
  },
});

// Receive messages
const response = await sqs.receiveMessage({
  QueueUrl: process.env.QUEUE_URL,
  MaxNumberOfMessages: 10,
  WaitTimeSeconds: 20, // Long polling
  MessageAttributeNames: ['All'],
});

for (const message of response.Messages || []) {
  await processMessage(message);

  // Delete after processing
  await sqs.deleteMessage({
    QueueUrl: process.env.QUEUE_URL,
    ReceiptHandle: message.ReceiptHandle,
  });
}

FIFO Queue#

await sqs.sendMessage({
  QueueUrl: process.env.FIFO_QUEUE_URL,
  MessageBody: JSON.stringify(payload),
  MessageGroupId: userId, // Orders from same user processed in order
  MessageDeduplicationId: orderId, // Prevent duplicates
});

Dead Letter Queues#

// RabbitMQ DLQ
await channel.assertQueue('orders', {
  durable: true,
  deadLetterExchange: 'dlx',
  deadLetterRoutingKey: 'orders-dlq',
});

// Reject to DLQ after retries
if (retryCount >= 3) {
  channel.nack(msg, false, false); // Don't requeue
} else {
  channel.nack(msg, false, true); // Requeue
}

Message Patterns#

Request-Reply#

// Request
const correlationId = crypto.randomUUID();
const replyQueue = await channel.assertQueue('', { exclusive: true });

channel.sendToQueue('rpc-queue', Buffer.from(request), {
  correlationId,
  replyTo: replyQueue.queue,
});

// Wait for reply
const reply = await new Promise((resolve) => {
  channel.consume(replyQueue.queue, (msg) => {
    if (msg?.properties.correlationId === correlationId) {
      resolve(JSON.parse(msg.content.toString()));
    }
  });
});

import { SNS } from '@aws-sdk/client-sns';

const sns = new SNS({});

// Publish to topic
await sns.publish({
  TopicArn: process.env.ORDERS_TOPIC_ARN,
  Message: JSON.stringify(order),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'order.created' },
  },
});

// Multiple SQS queues subscribe to the topic
// - notifications-queue
// - analytics-queue
// - inventory-queue

Best Practices#

Use persistent messages: Survive broker restarts
Implement idempotency: Handle duplicate messages
Set message TTL: Prevent queue buildup
Use dead letter queues: Capture failed messages
Monitor queue depth: Alert on growing backlogs
Graceful shutdown: Finish processing before stopping

Choose RabbitMQ for complex routing, SQS for AWS-native simplicity.

Share this article