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

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

docker compose up -d

# Verify Kafka is running
docker ps
# You should see a container named "kafka1"

# Clone the docker-compose repo
git clone https://github.com/conduktor/kafka-stack-docker-compose.git
cd kafka-stack-docker-compose

# Start single-broker Kafka + Conduktor web UI
docker compose -f conduktor-kafka-single.yml up -d

1. 1
   Clone the repository:

   git clone https://github.com/audoir/kafka-tutorial.git cd kafka-tutorial
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'],
});

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

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

// A 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 (broker: localhost:9092)

# If using the minimal setup
docker compose down

# If using conduktor
docker compose -f conduktor-kafka-single.yml down

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

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -v rabbitmq_data:/var/lib/rabbitmq \
  --hostname rabbitmq-host \
  rabbitmq:management

version: "3.8"
services:
  rabbitmq:
    image: rabbitmq:management
    container_name: rabbitmq
    ports:
      - "5672:5672"   # AMQP protocol port
      - "15672:15672" # Management UI port
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

volumes:
  rabbitmq_data:

docker compose up -d

# Check if RabbitMQ is running
docker ps | grep rabbitmq

# View logs
docker logs rabbitmq

# Stop RabbitMQ
docker stop rabbitmq

# Start again
docker start rabbitmq

# Remove container (data lost unless using volume)
docker rm -f rabbitmq

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

RABBITMQ_URL=amqp://username:password@your-host:5672

Why? In RabbitMQ, messages are routed to queues at the moment of publishing. If a queue doesn't exist yet (or isn't bound to the exchange), the message is silently dropped. The setup step ensures all queues and bindings exist before any messages are sent.

app/
├── page.tsx                    # Tutorial home page
├── api/
│   ├── status/route.ts         # GET: check RabbitMQ connection
│   ├── fanout/
│   │   ├── setup/route.ts      # POST: create exchange, queues, bindings
│   │   ├── publish/route.ts    # POST: publish to fanout exchange
│   │   └── consume/route.ts    # GET: consume from a named queue
│   ├── direct/
│   │   ├── setup/route.ts      # POST: create exchange, queues, bindings
│   │   ├── publish/route.ts    # POST: publish with routing key
│   │   └── consume/route.ts    # GET: consume by routing key
│   ├── topic/
│   │   ├── setup/route.ts      # POST: create exchange, queues, bindings
│   │   ├── publish/route.ts    # POST: publish with dot-separated key
│   │   └── consume/route.ts    # GET: consume with wildcard pattern
│   ├── headers/
│   │   ├── setup/route.ts      # POST: create exchange, queues, bindings
│   │   ├── publish/route.ts    # POST: publish with custom headers
│   │   └── consume/route.ts    # GET: consume with x-match filter
│   └── deadletter/
│       ├── setup/route.ts      # POST: create both exchanges, both queues, bindings
│       ├── publish/route.ts    # POST: publish to main queue (optional TTL)
│       ├── reject/route.ts     # POST: NACK a message → routes to DLX
│       └── consume/route.ts    # GET: consume from dead letter queue
└── tutorial/
    ├── fanout/page.tsx         # Fanout tutorial page
    ├── direct/page.tsx         # Direct tutorial page
    ├── topic/page.tsx          # Topic tutorial page
    ├── headers/page.tsx        # Headers tutorial page
    └── deadletter/page.tsx     # Dead letter queue tutorial page

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

// 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();
}

git clone https://github.com/audoir/sqs-sns-tutorial.git
cd sqs-sns-tutorial
npm install

AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=your_access_key_here
AWS_SECRET_ACCESS_KEY=your_secret_key_here
SQS_QUEUE_URL=https://sqs.us-east-1.amazonaws.com/123456789012/your-queue-name
SNS_TOPIC_ARN=arn:aws:sns:us-east-1:123456789012:your-topic-name

npm run dev

// 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);

// Send a message to SQS
const response = await fetch('/api/sqs/send', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ message: 'Hello World!' })
});

// Receive 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));

// Subscribe to topic (via API)
const subscription = await fetch('/api/sns/subscribe', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    protocol: 'email',
    endpoint: 'user@example.com'
  })
});

// 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
  }));
}

├── app/
│   ├── api/
│   │   ├── sqs/          # SQS API endpoints
│   │   │   ├── send/     # Send messages
│   │   │   ├── receive/  # Receive messages
│   │   │   ├── delete/   # Delete messages
│   │   │   └── status/   # Queue status
│   │   └── sns/          # SNS API endpoints
│   │       ├── publish/      # Publish messages
│   │       ├── subscribe/    # Create subscriptions
│   │       └── subscriptions/ # List subscriptions
│   ├── page.tsx          # Main demo interface
│   └── layout.tsx        # App layout
├── components/
│   ├── sqs/
│   │   ├── SQSTab.tsx           # Main SQS tab container
│   │   ├── SendMessage.tsx      # Message sending component
│   │   └── ReceiveMessages.tsx  # Message receiving component
│   ├── sns/
│   │   ├── SNSTab.tsx           # Main SNS tab container
│   │   ├── PublishMessage.tsx   # Message publishing component
│   │   └── ManageSubscriptions.tsx # Subscription management
│   ├── TabNavigation.tsx        # Tab switching component
│   └── SetupInstructions.tsx    # Setup instructions component
├── hooks/
│   ├── useSQS.ts        # SQS state management and API calls
│   └── useSNS.ts        # SNS state management and API calls
├── lib/
│   ├── aws-config.ts    # AWS SDK configuration
│   ├── sqs-service.ts   # SQS service wrapper
│   ├── sns-service.ts   # SNS service wrapper
│   └── types.ts         # TypeScript type definitions
├── AWS_SETUP.md         # Detailed AWS setup guide
└── env.example          # Environment variables template

// hooks/useSQS.ts — 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 };
}