Message Queues: RabbitMQ and Kafka for Async Processing

┌─────────────────────────────────────────────────────────────────┐
│                    RabbitMQ vs Kafka                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  RabbitMQ                                                       │
│  ─────────                                                      │
│  • Message broker (smart broker, dumb consumer)                │
│  • Messages are deleted after consumption                       │
│  • Complex routing (exchanges, bindings)                        │
│  • Better for task distribution                                 │
│  • Use: RPC, background jobs, request-reply patterns            │
│                                                                  │
│  Kafka                                                          │
│  ───────                                                        │
│  • Distributed event streaming platform (dumb broker, smart consumer)│
│  • Messages persisted for configurable time                     │
│  • Simple publish-subscribe with partitions                     │
│  • Better for event sourcing, stream processing                 │
│  • Use: Event-driven architecture, analytics, log aggregation   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

When to use which:
┌─────────────────────────────────────────────────────────────────┐
│ Use RabbitMQ when:                    Use Kafka when:           │
├─────────────────────────────────────────────────────────────────┤
│ • You need message routing          • You need event streaming  │
│ • Messages are transient            • Messages need replay      │
│ • Priority queues needed            • High throughput required  │
│ • Complex message patterns          • Event sourcing            │
│ • RPC over messaging                • Data pipelines            │
│ • Smaller team (easier ops)         • Large-scale team          │
└─────────────────────────────────────────────────────────────────┘

# docker-compose-rabbitmq.yml
version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3.12-management
    ports:
      - "5672:5672"   # AMQP protocol
      - "15672:15672" # Management UI
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 30s
      timeout: 10s
      retries: 5

volumes:
  rabbitmq-data:

import pika
import json
import threading
import time

# Producer - sends messages
class RabbitMQProducer:
    def __init__(self, host='localhost'):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host))
        self.channel = self.connection.channel()
        # Declare queue (idempotent)
        self.channel.queue_declare(queue='task_queue', durable=True)
        # Declare exchange for routing
        self.channel.exchange_declare(exchange='events', exchange_type='topic', durable=True)

    def send_task(self, task_type, payload):
        message = json.dumps({
            'type': task_type,
            'payload': payload,
            'timestamp': time.time()
        })
        self.channel.basic_publish(
            exchange='',
            routing_key='task_queue',
            body=message,
            properties=pika.BasicProperties(
                delivery_mode=2,  # make message persistent
                priority=task_type == 'urgent' and 10 or 5
            )
        )
        print(f"Sent {task_type}: {payload}")

    def publish_event(self, event_type, data):
        message = json.dumps({
            'event_type': event_type,
            'data': data,
            'timestamp': time.time()
        })
        self.channel.basic_publish(
            exchange='events',
            routing_key=event_type,
            body=message
        )

# Consumer - processes messages
class RabbitMQConsumer:
    def __init__(self, host='localhost', queue='task_queue'):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=queue, durable=True)
        # Only process one message at a time
        self.channel.basic_qos(prefetch_count=1)

    def start(self, callback):
        self.channel.basic_consume(queue='task_queue', on_message_callback=callback)
        print('Waiting for messages...')
        self.channel.start_consuming()

# Example consumer callback
def process_message(ch, method, properties, body):
    message = json.loads(body)
    print(f"Processing: {message['type']}")

    # Simulate work
    time.sleep(1)

    # Acknowledge message (remove from queue)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(f"Completed: {message['type']}")

# Dead Letter Queue for failed messages
class DeadLetterHandler:
    def __init__(self, host='localhost'):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host))
        self.channel = self.connection.channel()

        # Setup dead letter exchange
        self.channel.exchange_declare(exchange='dlx', exchange_type='direct')
        self.channel.queue_declare(queue='dead_letter_queue', durable=True)
        self.channel.queue_bind(queue='dead_letter_queue', exchange='dlx', routing_key='failed')

    def send_to_dlq(self, message, error):
        """Send failed message to dead letter queue"""
        failed_message = {
            'original_message': json.loads(message),
            'error': str(error),
            'failed_at': time.time()
        }
        self.channel.basic_publish(
            exchange='dlx',
            routing_key='failed',
            body=json.dumps(failed_message),
            properties=pika.BasicProperties(delivery_mode=2)
        )

# docker-compose-kafka.yml
version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092

from kafka import KafkaProducer, KafkaConsumer
import json
import time

# Kafka Producer
class KafkaEventProducer:
    def __init__(self, bootstrap_servers='localhost:9092'):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # Wait for all replicas
            retries=3,
            compression_type='gzip'
        )

    def send_event(self, topic, event_type, data, key=None):
        """Send event to Kafka topic"""
        message = {
            'event_type': event_type,
            'data': data,
            'timestamp': int(time.time()),
            'source': 'order-service'
        }
        # Key ensures same partition (ordering per key)
        key_bytes = key.encode('utf-8') if key else None
        future = self.producer.send(topic, key=key_bytes, value=message)

        # Wait for acknowledgment (optional, for critical events)
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic} partition {record_metadata.partition}")
        return record_metadata

    def send_batch(self, topic, events):
        """Send multiple events in a batch"""
        with self.producer as batch_producer:
            for event in events:
                batch_producer.send(topic, value=event)

# Kafka Consumer
class KafkaEventConsumer:
    def __init__(self, bootstrap_servers='localhost:9092', group_id='order-group'):
        self.consumer = KafkaConsumer(
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            auto_offset_reset='earliest',  # Start from earliest when no offset
            enable_auto_commit=False,      # Manual commit for processing guarantees
            max_poll_records=100
        )

    def subscribe(self, topics):
        self.consumer.subscribe(topics)

    def consume(self, handler):
        """Consume messages and process with handler"""
        try:
            for message in self.consumer:
                try:
                    handler(message.value)
                    self.consumer.commit()  # Commit after successful processing
                except Exception as e:
                    print(f"Error processing message: {e}")
                    # Don't commit; message will be replayed
                    # Could send to dead letter queue after N retries
        except KeyboardInterrupt:
            self.consumer.close()

# Usage examples
producer = KafkaEventProducer()

# Send order created event (key by order_id for ordering)
producer.send_event('orders', 'order.created', {
    'order_id': 12345,
    'user_id': 789,
    'total': 99.99,
    'items': [{'product_id': 1, 'quantity': 2}]
}, key='12345')

# Send payment processed event
producer.send_event('payments', 'payment.succeeded', {
    'payment_id': 'pay_abc123',
    'order_id': 12345,
    'amount': 99.99
})

# Consumer example
def handle_order_event(event):
    print(f"Received: {event['event_type']}")
    if event['event_type'] == 'order.created':
        # Process order
        update_inventory(event['data']['items'])
        send_confirmation_email(event['data']['user_id'])

consumer = KafkaEventConsumer(group_id='order-processor')
consumer.subscribe(['orders', 'payments'])
consumer.consume(handle_order_event)

# event-driven-order-system.py
import json
import time
import threading
from datetime import datetime

# Order Service - Producer
class OrderService:
    def __init__(self, producer):
        self.producer = producer

    def create_order(self, user_id, items):
        order_id = generate_order_id()

        # Calculate total
        total = sum(item['price'] * item['quantity'] for item in items)

        # Create order record
        order = {
            'order_id': order_id,
            'user_id': user_id,
            'items': items,
            'total': total,
            'status': 'pending',
            'created_at': time.time()
        }

        # Save to database
        save_order_to_db(order)

        # Publish events
        self.producer.send_event('orders', 'order.created', order, key=str(order_id))

        return order

# Payment Service - Consumer and Producer
class PaymentService:
    def __init__(self, producer, consumer):
        self.producer = producer
        self.consumer = consumer
        self.consumer.subscribe(['orders'])

    def start(self):
        def handle(event):
            if event['event_type'] == 'order.created':
                self.process_payment(event['data'])

        self.consumer.consume(handle)

    def process_payment(self, order):
        # Process payment (simulate external API)
        try:
            payment_result = charge_credit_card(order['user_id'], order['total'])

            # Publish payment result
            event = {
                'order_id': order['order_id'],
                'payment_id': payment_result['payment_id'],
                'amount': order['total'],
                'status': 'succeeded'
            }
            self.producer.send_event('payments', 'payment.succeeded', event, key=str(order['order_id']))

        except Exception as e:
            # Publish payment failure
            event = {
                'order_id': order['order_id'],
                'error': str(e),
                'amount': order['total']
            }
            self.producer.send_event('payments', 'payment.failed', event, key=str(order['order_id']))

# Inventory Service
class InventoryService:
    def __init__(self, producer, consumer):
        self.producer = producer
        self.consumer = consumer
        self.consumer.subscribe(['payments'])

    def start(self):
        def handle(event):
            if event['event_type'] == 'payment.succeeded':
                self.reserve_inventory(event['data'])

        self.consumer.consume(handle)

    def reserve_inventory(self, payment):
        # Reserve items in inventory
        update_inventory(payment['order_id'])

        # Publish inventory reserved
        self.producer.send_event('inventory', 'inventory.reserved', payment)

# Shipping Service
class ShippingService:
    def __init__(self, consumer):
        self.consumer = consumer
        self.consumer.subscribe(['inventory', 'payments'])

    def start(self):
        def handle(event):
            if event['event_type'] == 'inventory.reserved':
                self.create_shipment(event['data'])
            elif event['event_type'] == 'payment.failed':
                self.handle_payment_failure(event['data'])

        self.consumer.consume(handle)

    def create_shipment(self, order):
        print(f"Creating shipment for order {order['order_id']}")
        # Create shipping label
        # Notify warehouse
        # Send tracking to customer

    def handle_payment_failure(self, failure):
        print(f"Payment failed for order {failure['order_id']}: {failure['error']}")
        # Notify customer
        # Update order status

# Notification Service
class NotificationService:
    def __init__(self, consumer):
        self.consumer = consumer
        self.consumer.subscribe(['orders', 'payments', 'shipping'])

    def start(self):
        def handle(event):
            if event['event_type'] == 'order.created':
                self.send_order_confirmation(event['data'])
            elif event['event_type'] == 'payment.succeeded':
                self.send_payment_confirmation(event['data'])
            elif event['event_type'] == 'shipping.created':
                self.send_tracking(event['data'])

        self.consumer.consume(handle)

    def send_order_confirmation(self, order):
        print(f"Sending order confirmation email to user {order['user_id']}")

    def send_payment_confirmation(self, payment):
        print(f"Sending payment confirmation for order {payment['order_id']}")

    def send_tracking(self, shipment):
        print(f"Sending tracking for order {shipment['order_id']}")

# Run all services
if __name__ == '__main__':
    # In production, each service would run in separate process/container
    producer = KafkaEventProducer()
    consumer = KafkaEventConsumer(group_id='order-processor')

    # Start services in threads
    threading.Thread(target=PaymentService(producer, consumer).start).start()
    threading.Thread(target=InventoryService(producer, consumer).start).start()
    threading.Thread(target=ShippingService(consumer).start).start()
    threading.Thread(target=NotificationService(consumer).start).start()

    # Order service handles API requests
    order_service = OrderService(producer)
    order_service.create_order(user_id=123, items=[{'product_id': 1, 'price': 99.99, 'quantity': 1}])