Code snippets — Message Queues Kafka Rabbitmq

snippet-2.go

// All events for user 42 go to the same partition
kafka.Message{
    Key:   []byte("user-42"),  // Hashed to determine partition
    Value: eventData,
}

snippet-2.py

# RabbitMQ DLQ
channel.queue_declare(
    queue='email_jobs',
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'email_jobs.failed',
        'x-message-ttl': 3600000  # 1 hour
    }
)

snippet-2.txt

Topic: user-events (3 partitions)
├── Partition 0 → Consumer A (in group "analytics")
├── Partition 1 → Consumer B (in group "analytics")
└── Partition 2 → Consumer C (in group "analytics")

Same topic, different group:
├── Partition 0,1,2 → Consumer X (in group "audit-log")

snippet-3.txt

Is your primary use case:

Background jobs / task distribution?
  → RabbitMQ

Real-time event streaming / high throughput?
  → Kafka

Multiple independent consumers of same data?
  → Kafka

Complex routing (topic, header, fanout)?
  → RabbitMQ

Need to replay events / audit history?
  → Kafka

Simple pub/sub at moderate scale?
  → Either (RabbitMQ simpler to operate)

snippet.go

// Kafka: Event producer
writer := kafka.NewWriter(kafka.WriterConfig{
    Brokers: []string{"localhost:9092"},
    Topic:   "user-events",
    Balancer: &kafka.LeastBytes{},
})

err := writer.WriteMessages(ctx, kafka.Message{
    Key:   []byte(userID),
    Value: json.Marshal(UserLoggedInEvent{
        UserID:    userID,
        Timestamp: time.Now(),
        IPAddress: ip,
    }),
})

// Kafka: Consumer group
reader := kafka.NewReader(kafka.ReaderConfig{
    Brokers:  []string{"localhost:9092"},
    Topic:    "user-events",
    GroupID:  "analytics-service",
    MinBytes: 10e3,
    MaxBytes: 10e6,
})

for {
    msg, err := reader.ReadMessage(ctx)
    if err != nil {
        break
    }
    processEvent(msg.Value)
}

snippet.py

# RabbitMQ: Simple work queue
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='email_jobs', durable=True)

# Producer
channel.basic_publish(
    exchange='',
    routing_key='email_jobs',
    body=json.dumps({'to': 'user@example.com', 'template': 'welcome'}),
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent
)

# Consumer
def process_email(ch, method, properties, body):
    job = json.loads(body)
    send_email(job['to'], job['template'])
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='email_jobs', on_message_callback=process_email)
channel.start_consuming()

snippet.sh

# Reset consumer group to beginning to replay all events
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic user-events \
  --reset-offsets --to-earliest --execute

snippet.txt

RabbitMQ:
Producer → Queue → Consumer (message deleted after ack)

Kafka:
Producer → Topic/Partition → Consumer Group A (reads at offset X)
                           → Consumer Group B (reads at offset Y)
                           (messages retained for N days)