Messaging Systems Compared: Kafka, RabbitMQ, AWS SQS & SNS

# docker-compose.yml — minimal single-broker Kafka (KRaft mode)
version: "3"
services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    container_name: kafka1
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

# Start Kafka
docker compose up -d

1. 1
   Clone the repository:

   git clone https://github.com/audoir/kafka-tutorial.git
2. 2
   Install dependencies:

   npm install
3. 3
   Start the development server:

   npm run dev

// lib/kafka.ts — The KafkaJS Client singleton
import { Kafka } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'kafka-tutorial',
  brokers: ['localhost:9092'],
});

// Shared admin client — used for topic management and cluster inspection
export const admin = kafka.admin();

// Shared producer — used to send messages to topics
export const producer = kafka.producer();

// Factory function — creates a new consumer with a given group ID
export function createConsumer(groupId: string) {
  return kafka.consumer({ groupId });
}

kafka-tutorial/
├── app/
│   ├── api/
│   │   ├── health/route.ts          # GET — ping Kafka, return broker/cluster info
│   │   ├── topics/route.ts          # GET list, POST create, DELETE topic
│   │   ├── produce/route.ts         # POST send messages
│   │   ├── consume/route.ts         # POST consume messages
│   │   └── consumer-groups/route.ts # GET list groups
│   ├── layout.tsx
│   └── page.tsx                     # Main tabbed UI + connection status badge
├── components/
│   ├── ConceptsPanel.tsx            # Theory reference with diagrams
│   ├── TopicsPanel.tsx              # Create/list/delete topics
│   ├── ProducerPanel.tsx            # Send messages with/without keys
│   ├── ConsumerPanel.tsx            # Consume and inspect messages
│   └── ConsumerGroupsPanel.tsx      # Group scenarios + live group list
└── lib/
    └── kafka.ts                     # KafkaJS client singleton

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  rabbitmq:management

1. 1
   Clone the repository:

   git clone https://github.com/audoir/rabbitmq-tutorial.git
2. 2
   Install dependencies:

   npm install
3. 3
   Start the development server:

   npm run dev

// Fanout: one publish → all queues receive a copy
channel.publish('fanout.exchange', '', Buffer.from(message));
// Routing key is empty string — it's ignored by fanout exchanges

// Direct: exact routing key match
channel.publish('direct.exchange', 'error', Buffer.from(message));
// Only queues bound with key 'error' receive this message

// Topic: wildcard pattern matching
// Pattern 'logs.*' matches 'logs.error' but NOT 'logs.error.critical'
// Pattern 'logs.#' matches both 'logs.error' and 'logs.error.critical'

channel.bindQueue('queue.star', 'topic.exchange', 'logs.*');
channel.bindQueue('queue.hash', 'topic.exchange', 'logs.#');

// Headers: attribute-based routing
channel.publish('headers.exchange', '', Buffer.from(message), {
  headers: { format: 'pdf', type: 'report' }
});

// Queue bound with x-match=all, format=pdf, type=report → MATCH ✓
// Queue bound with x-match=all, format=pdf, type=invoice → NO MATCH ✗
// Queue bound with x-match=any, format=pdf, type=invoice → MATCH ✓

// Setup: main queue with dead letter exchange configured
channel.assertQueue('main.queue', {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'dlx.exchange',
    'x-message-ttl': 5000  // optional: auto-expire after 5s
  }
});

// Reject a message → routes to DLX → lands in dead letter queue
channel.nack(msg, false, false);  // requeue=false → dead letter

// x-death header added by RabbitMQ
{
  queue: 'main.queue',        // original queue name
  reason: 'rejected',         // 'rejected', 'expired', or 'maxlen'
  time: <timestamp>,
  exchange: 'main.exchange',
  'routing-keys': ['key']
}

// lib/rabbitmq.ts
import amqp from 'amqplib';

const RABBITMQ_URL = process.env.RABBITMQ_URL || 'amqp://guest:guest@localhost:5672';

export async function createConnection() {
  const connection = await amqp.connect(RABBITMQ_URL);
  const channel = await connection.createChannel();
  return { connection, channel };
}

export async function closeConnection(
  channel: amqp.Channel,
  connection: amqp.Connection
) {
  await channel.close();
  await connection.close();
}

app/
├── page.tsx                    # Tutorial home page
├── api/
│   ├── status/route.ts         # GET: check RabbitMQ connection
│   ├── fanout/
│   │   ├── setup/route.ts
│   │   ├── publish/route.ts
│   │   └── consume/route.ts
│   ├── direct/
│   │   ├── setup/route.ts
│   │   ├── publish/route.ts
│   │   └── consume/route.ts
│   ├── topic/
│   │   ├── setup/route.ts
│   │   ├── publish/route.ts
│   │   └── consume/route.ts
│   ├── headers/
│   │   ├── setup/route.ts
│   │   ├── publish/route.ts
│   │   └── consume/route.ts
│   └── deadletter/
│       ├── setup/route.ts
│       ├── publish/route.ts
│       ├── reject/route.ts
│       └── consume/route.ts
└── tutorial/
    ├── fanout/page.tsx
    ├── direct/page.tsx
    ├── topic/page.tsx
    ├── headers/page.tsx
    └── deadletter/page.tsx

lib/
└── rabbitmq.ts                 # Shared: createConnection / closeConnection

// AWS SDK v3 Configuration
import { SQSClient } from '@aws-sdk/client-sqs';
import { SNSClient } from '@aws-sdk/client-sns';

const awsConfig = {
  region: process.env.AWS_REGION || 'us-east-1',
  credentials: {
    accessKeyId: process.env.AWS_ACCESS_KEY_ID!,
    secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY!,
  },
};

export const sqsClient = new SQSClient(awsConfig);
export const snsClient = new SNSClient(awsConfig);

// Sending a message to SQS
const params = {
  QueueUrl: process.env.SQS_QUEUE_URL,
  MessageBody: JSON.stringify({
    orderId: '12345',
    customerId: 'user-456',
    items: ['product-1', 'product-2'],
    timestamp: new Date().toISOString()
  }),
  MessageAttributes: {
    'MessageType': {
      DataType: 'String',
      StringValue: 'ORDER_CREATED'
    }
  }
};

const result = await sqsClient.send(new SendMessageCommand(params));

// Receiving messages with long polling
const receiveParams = {
  QueueUrl: process.env.SQS_QUEUE_URL,
  MaxNumberOfMessages: 10,
  WaitTimeSeconds: 20, // Long polling
  MessageAttributeNames: ['All']
};

const messages = await sqsClient.send(
  new ReceiveMessageCommand(receiveParams)
);

// Process each message
for (const message of messages.Messages || []) {
  try {
    await processMessage(JSON.parse(message.Body));
    
    // Delete the message after successful processing
    await sqsClient.send(new DeleteMessageCommand({
      QueueUrl: process.env.SQS_QUEUE_URL,
      ReceiptHandle: message.ReceiptHandle
    }));
  } catch (error) {
    console.error('Message processing failed:', error);
    // Message will become visible again after visibility timeout
  }
}

// Publishing a message to SNS topic
const publishParams = {
  TopicArn: process.env.SNS_TOPIC_ARN,
  Message: JSON.stringify({
    event: 'USER_REGISTERED',
    userId: 'user-789',
    email: 'user@example.com',
    timestamp: new Date().toISOString()
  }),
  MessageAttributes: {
    'event_type': {
      DataType: 'String',
      StringValue: 'USER_REGISTERED'
    },
    'priority': {
      DataType: 'String',
      StringValue: 'high'
    }
  }
};

const result = await snsClient.send(new PublishCommand(publishParams));

// Creating subscriptions for different protocols
const subscriptions = [
  {
    Protocol: 'email',
    Endpoint: 'admin@example.com'
  },
  {
    Protocol: 'sqs',
    Endpoint: 'arn:aws:sqs:us-east-1:123456789012:notification-queue'
  },
  {
    Protocol: 'https',
    Endpoint: 'https://api.example.com/webhooks/notifications'
  }
];

for (const subscription of subscriptions) {
  await snsClient.send(new SubscribeCommand({
    TopicArn: process.env.SNS_TOPIC_ARN,
    Protocol: subscription.Protocol,
    Endpoint: subscription.Endpoint
  }));
}

// Custom hook for SQS operations
export function useSQS() {
  const [messages, setMessages] = useState<SQSMessage[]>([]);
  const [loading, setLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);

  const sendMessage = async (messageBody: string) => {
    setLoading(true);
    setError(null);
    
    try {
      const response = await fetch('/api/sqs/send', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ message: messageBody }),
      });
      
      if (!response.ok) throw new Error('Failed to send message');
      return await response.json();
    } catch (err) {
      setError(err instanceof Error ? err.message : 'Unknown error');
      throw err;
    } finally {
      setLoading(false);
    }
  };

  return { messages, loading, error, sendMessage };
}

1. 1
   Clone the repository:

   git clone https://github.com/audoir/sqs-sns-tutorial.git
2. 2
   Install dependencies:

   npm install
3. 3
   Configure AWS services:

   Follow AWS_SETUP.md for detailed instructions
4. 4
   Set up environment variables:

   Copy .env.example to .env.local and configure
5. 5
   Start the development server:

   npm run dev