Message Queues Fundamentals

The Synchronous Problem

Synchronous systems: Service A calls Service B and waits for response.

Problems:

• ❌ Service A blocks waiting for B
• ❌ If B is slow, A is slow
• ❌ If B crashes, A fails
• ❌ Tight coupling

Solution: Message Queues - Asynchronous, decoupled communication!

---

What is a Message Queue?

Message queue = Buffer that stores messages between senders and receivers.

Simple Analogy

Think of a message queue like a mailbox:

• Producer = Person sending mail
• Queue = Mailbox (stores mail)
• Consumer = Mail carrier (picks up mail)

The sender doesn’t wait for the carrier. The mailbox decouples them!

Key Benefits

1. Decoupling - Services don’t know about each other
2. Asynchronous - Non-blocking communication
3. Reliability - Messages persisted, can retry
4. Scalability - Multiple consumers, load distribution
5. Resilience - If consumer fails, messages stay in queue

---

Pattern 1: Producer-Consumer

One producer, one or more consumers. Each message processed once.

Characteristics:

• ✅ Load balancing - Multiple consumers share work
• ✅ Task distribution - Work distributed across consumers
• ✅ Scalability - Add more consumers to scale
• ✅ Each message processed once - One consumer gets each message

Use cases:

• Background job processing
• Email sending
• Image processing
• Data transformation
• Task queues

Producer-Consumer Implementation

Python

"producer_consumer.py

import queue
import threading
import time
from typing import Optional

class MessageQueue:
    """Simple message queue implementation"""

    def __init__(self, maxsize: int = 1000):
        self.queue = queue.Queue(maxsize=maxsize)
        self.consumers = []
        self.running = False

    def produce(self, message: dict):
        """Producer: Add message to queue"""
        try:
            self.queue.put_nowait(message)
            print(f"Produced: {message}")
        except queue.Full:
            print("Queue full! Message rejected.")

    def consume(self, handler):
        """Consumer: Process messages from queue"""
        while self.running:
            try:
                message = self.queue.get(timeout=1)
                # Process message
                handler(message)
                # Acknowledge
                self.queue.task_done()
            except queue.Empty:
                continue

    def start_consumer(self, handler, consumer_id: int):
        """Start a consumer thread"""
        def consumer_loop():
            print(f"Consumer {consumer_id} started")
            self.consume(handler)

        thread = threading.Thread(target=consumer_loop, daemon=True)
        thread.start()
        self.consumers.append(thread)

    def start(self):
        """Start queue processing"""
        self.running = True

    def stop(self):
        """Stop queue processing"""
        self.running = False
        self.queue.join()  # Wait for all tasks to complete

# Usage
queue = MessageQueue(maxsize=100)
queue.start()

# Message handler
def process_order(message):
    print(f"Processing order: {message['order_id']}")
    # Process order...
    time.sleep(1)  # Simulate work
    print(f"Order {message['order_id']} processed")

# Start multiple consumers
queue.start_consumer(process_order, consumer_id=1)
queue.start_consumer(process_order, consumer_id=2)

# Producer sends messages
for i in range(10):
    queue.produce({'order_id': i, 'amount': 100 + i})
    time.sleep(0.1)

time.sleep(5)  # Let consumers process
queue.stop()

Java

"ProducerConsumer.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

public class MessageQueue {
    private final BlockingQueue<Message> queue;
    private final ExecutorService executor;
    private volatile boolean running = false;

    public MessageQueue(int maxSize) {
        this.queue = new LinkedBlockingQueue<>(maxSize);
        this.executor = Executors.newCachedThreadPool();
    }

    public void produce(Message message) {
        // Producer: Add message to queue
        if (queue.offer(message)) {
            System.out.println("Produced: " + message);
        } else {
            System.out.println("Queue full! Message rejected.");
        }
    }

    public void consume(Consumer<Message> handler) {
        // Consumer: Process messages from queue
        while (running) {
            try {
                Message message = queue.take();  // Blocks until message available
                handler.accept(message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    public void startConsumer(Consumer<Message> handler, int consumerId) {
        // Start a consumer thread
        executor.submit(() -> {
            System.out.println("Consumer " + consumerId + " started");
            consume(handler);
        });
    }

    public void start() {
        running = true;
    }

    public void stop() {
        running = false;
        executor.shutdown();
    }
}

// Usage
MessageQueue queue = new MessageQueue(1000);
queue.start();

// Message handler
Consumer<Message> processOrder = message -> {
    System.out.println("Processing order: " + message.getOrderId());
    // Process order...
    try {
        Thread.sleep(1000);  // Simulate work
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    System.out.println("Order " + message.getOrderId() + " processed");
};

// Start multiple consumers
queue.startConsumer(processOrder, 1);
queue.startConsumer(processOrder, 2);

// Producer sends messages
for (int i = 0; i < 10; i++) {
    queue.produce(new Message(i, 100 + i));
    Thread.sleep(100);
}

Thread.sleep(5000);  // Let consumers process
queue.stop();

---

Pattern 2: Pub-Sub (Publish-Subscribe)

One publisher, multiple subscribers. Each subscriber gets copy of message.

Characteristics:

• ✅ Fan-out - One message → multiple consumers
• ✅ Decoupling - Publisher doesn’t know subscribers
• ✅ Event broadcasting - Notify multiple services
• ✅ Each subscriber gets copy - Independent processing

Use cases:

• Event notifications
• Real-time updates
• Cache invalidation
• Log aggregation
• Analytics events

Pub-Sub Implementation

Python

"pub_sub.py

from typing import List, Callable, Dict, Any
from threading import Lock
import threading

class Topic:
    """Pub-sub topic"""

    def __init__(self, name: str):
        self.name = name
        self.subscribers: List[Callable] = []
        self.lock = Lock()

    def subscribe(self, handler: Callable):
        """Subscribe to topic"""
        with self.lock:
            self.subscribers.append(handler)
            print(f"Subscriber added to {self.name}. Total: {len(self.subscribers)}")

    def unsubscribe(self, handler: Callable):
        """Unsubscribe from topic"""
        with self.lock:
            if handler in self.subscribers:
                self.subscribers.remove(handler)

    def publish(self, message: Dict[str, Any]):
        """Publish message to all subscribers"""
        with self.lock:
            subscribers = self.subscribers.copy()

        # Notify all subscribers (asynchronously)
        for subscriber in subscribers:
            try:
                # Run in separate thread to avoid blocking
                threading.Thread(
                    target=subscriber,
                    args=(message,),
                    daemon=True
                ).start()
            except Exception as e:
                print(f"Error notifying subscriber: {e}")

class PubSubBroker:
    """Pub-sub message broker"""

    def __init__(self):
        self.topics: Dict[str, Topic] = {}
        self.lock = Lock()

    def get_topic(self, name: str) -> Topic:
        """Get or create topic"""
        with self.lock:
            if name not in self.topics:
                self.topics[name] = Topic(name)
            return self.topics[name]

    def publish(self, topic_name: str, message: Dict[str, Any]):
        """Publish message to topic"""
        topic = self.get_topic(topic_name)
        topic.publish(message)

    def subscribe(self, topic_name: str, handler: Callable):
        """Subscribe to topic"""
        topic = self.get_topic(topic_name)
        topic.subscribe(handler)

# Usage
broker = PubSubBroker()

# Subscribers
def email_handler(message):
    print(f"Email Service: Sending email for {message['event']}")

def sms_handler(message):
    print(f"SMS Service: Sending SMS for {message['event']}")

def analytics_handler(message):
    print(f"Analytics Service: Recording {message['event']}")

# Subscribe to 'user.created' topic
broker.subscribe('user.created', email_handler)
broker.subscribe('user.created', sms_handler)
broker.subscribe('user.created', analytics_handler)

# Publisher publishes event
broker.publish('user.created', {
    'event': 'user.created',
    'user_id': 123,
    'email': 'john@example.com'
})

Java

"PubSub.java

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

class Topic {
    private final String name;
    private final List<Consumer<Message>> subscribers = new CopyOnWriteArrayList<>();

    public Topic(String name) {
        this.name = name;
    }

    public void subscribe(Consumer<Message> handler) {
        subscribers.add(handler);
        System.out.println("Subscriber added to " + name + ". Total: " + subscribers.size());
    }

    public void unsubscribe(Consumer<Message> handler) {
        subscribers.remove(handler);
    }

    public void publish(Message message) {
        // Notify all subscribers
        subscribers.forEach(subscriber -> {
            try {
                subscriber.accept(message);
            } catch (Exception e) {
                System.err.println("Error notifying subscriber: " + e.getMessage());
            }
        });
    }
}

class PubSubBroker {
    private final Map<String, Topic> topics = new HashMap<>();

    public synchronized Topic getTopic(String name) {
        return topics.computeIfAbsent(name, Topic::new);
    }

    public void publish(String topicName, Message message) {
        Topic topic = getTopic(topicName);
        topic.publish(message);
    }

    public void subscribe(String topicName, Consumer<Message> handler) {
        Topic topic = getTopic(topicName);
        topic.subscribe(handler);
    }
}

// Usage
PubSubBroker broker = new PubSubBroker();

// Subscribers
Consumer<Message> emailHandler = message ->
    System.out.println("Email Service: Sending email for " + message.getEvent());

Consumer<Message> smsHandler = message ->
    System.out.println("SMS Service: Sending SMS for " + message.getEvent());

Consumer<Message> analyticsHandler = message ->
    System.out.println("Analytics Service: Recording " + message.getEvent());

// Subscribe to 'user.created' topic
broker.subscribe("user.created", emailHandler);
broker.subscribe("user.created", smsHandler);
broker.subscribe("user.created", analyticsHandler);

// Publisher publishes event
broker.publish("user.created", new Message("user.created", 123, "john@example.com"));

---

Message Delivery Guarantees

At-Most-Once (Fire and Forget)

Message may be lost, but never duplicated.

Characteristics:

• ✅ Simple
• ✅ Low latency
• ❌ May lose messages
• ❌ No retry

Use when: Non-critical messages, metrics, logs

At-Least-Once (Guaranteed Delivery)

Message delivered at least once, may have duplicates.

Characteristics:

• ✅ Reliable (won’t lose messages)
• ✅ Retries on failure
• ❌ May have duplicates
• ❌ Consumer must be idempotent

Use when: Critical messages, order processing, payments

Exactly-Once (Hardest)

Message delivered exactly once. Requires deduplication.

Characteristics:

• ✅ No duplicates
• ✅ No lost messages
• ❌ Complex implementation
• ❌ Performance overhead

Use when: Financial transactions, critical operations

---

Message Ordering

Ordering ensures messages processed in sequence.

Why Ordering Matters

Example: User account balance updates

Message 1: Balance = 100
Message 2: Balance = 150 (add 50)
Message 3: Balance = 120 (subtract 30)

Correct order: 100 → 150 → 120 ✅
Wrong order: 100 → 120 → 150 = 150 ❌ (wrong!)

Ordering Strategies

1. Per-Partition Ordering (Kafka)

• Messages in same partition processed in order
• Different partitions can process in parallel

2. Per-Queue Ordering (RabbitMQ)

• Single consumer per queue
• Messages processed sequentially

3. Global Ordering

• All messages processed in order
• Slower (no parallelism)

---

When to Use Message Queues

Use Queues When:

• ✅ Decoupling needed - Services shouldn’t know about each other
• ✅ Async processing - Don’t want to wait for response
• ✅ Load leveling - Smooth out traffic spikes
• ✅ Reliability - Need guaranteed delivery
• ✅ Scalability - Need to scale consumers independently

Don’t Use Queues When:

• ❌ Synchronous operations - Need immediate response
• ❌ Simple request-response - Overhead not worth it
• ❌ Ordering not needed - Can use simpler patterns
• ❌ Low latency critical - Queues add latency

Start Simple

Start with synchronous calls. Add queues only when you have clear evidence of: decoupling needs, async requirements, or scalability issues.

---

LLD Connection: Implementing Message Handlers

At the code level, message queues translate to producer/consumer classes, message handlers, and acknowledgment logic.

Message Handler Design

Python

"message_handler.py

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import json

class MessageHandler(ABC):
    """Base message handler interface"""

    @abstractmethod
    def handle(self, message: Dict[str, Any]) -> bool:
        """
        Handle message. Returns True if successful.
        Should be idempotent for at-least-once delivery.
        """
        pass

    @abstractmethod
    def can_handle(self, message_type: str) -> bool:
        """Check if handler can process this message type"""
        pass

class OrderProcessor(MessageHandler):
    """Process order messages"""

    def __init__(self, order_service):
        self.order_service = order_service
        self.processed_ids = set()  # For idempotency

    def can_handle(self, message_type: str) -> bool:
        return message_type == 'order.created'

    def handle(self, message: Dict[str, Any]) -> bool:
        order_id = message.get('order_id')

        # Idempotency check
        if order_id in self.processed_ids:
            print(f"Order {order_id} already processed. Skipping.")
            return True  # Already processed, consider success

        try:
            # Process order
            self.order_service.process_order(order_id, message)

            # Mark as processed
            self.processed_ids.add(order_id)

            return True
        except Exception as e:
            print(f"Error processing order {order_id}: {e}")
            return False  # Return False to trigger retry

class MessageConsumer:
    """Consumer that routes messages to handlers"""

    def __init__(self, queue, handlers: List[MessageHandler]):
        self.queue = queue
        self.handlers = handlers

    def consume(self):
        """Consume messages from queue"""
        while True:
            try:
                message_data = self.queue.get(timeout=1)
                message = json.loads(message_data)

                # Find handler
                handler = self.find_handler(message.get('type'))

                if handler:
                    # Process message
                    success = handler.handle(message)

                    if success:
                        # Acknowledge message
                        self.queue.task_done()
                    else:
                        # Return to queue for retry
                        self.queue.put(message_data)
                else:
                    print(f"No handler for message type: {message.get('type')}")
                    self.queue.task_done()

            except Exception as e:
                print(f"Error consuming message: {e}")

    def find_handler(self, message_type: str) -> Optional[MessageHandler]:
        """Find handler for message type"""
        for handler in self.handlers:
            if handler.can_handle(message_type):
                return handler
        return None

Java

"MessageHandler.java

import java.util.*;

interface MessageHandler {
    boolean handle(Message message);
    boolean canHandle(String messageType);
}

class OrderProcessor implements MessageHandler {
    private final OrderService orderService;
    private final Set<String> processedIds = new HashSet<>();

    public OrderProcessor(OrderService orderService) {
        this.orderService = orderService;
    }

    @Override
    public boolean canHandle(String messageType) {
        return "order.created".equals(messageType);
    }

    @Override
    public boolean handle(Message message) {
        String orderId = message.getOrderId();

        // Idempotency check
        synchronized (processedIds) {
            if (processedIds.contains(orderId)) {
                System.out.println("Order " + orderId + " already processed. Skipping.");
                return true;  // Already processed
            }
        }

        try {
            // Process order
            orderService.processOrder(orderId, message);

            // Mark as processed
            synchronized (processedIds) {
                processedIds.add(orderId);
            }

            return true;
        } catch (Exception e) {
            System.err.println("Error processing order " + orderId + ": " + e.getMessage());
            return false;  // Return false to trigger retry
        }
    }
}

class MessageConsumer {
    private final BlockingQueue<String> queue;
    private final List<MessageHandler> handlers;

    public MessageConsumer(BlockingQueue<String> queue, List<MessageHandler> handlers) {
        this.queue = queue;
        this.handlers = handlers;
    }

    public void consume() {
        while (true) {
            try {
                String messageData = queue.take();
                Message message = parseMessage(messageData);

                // Find handler
                MessageHandler handler = findHandler(message.getType());

                if (handler != null) {
                    // Process message
                    boolean success = handler.handle(message);

                    if (success) {
                        // Message processed successfully
                        // (Acknowledgment handled by queue)
                    } else {
                        // Return to queue for retry
                        queue.put(messageData);
                    }
                } else {
                    System.err.println("No handler for message type: " + message.getType());
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                System.err.println("Error consuming message: " + e.getMessage());
            }
        }
    }

    private MessageHandler findHandler(String messageType) {
        return handlers.stream()
            .filter(h -> h.canHandle(messageType))
            .findFirst()
            .orElse(null);
    }
}

---

Key Takeaways

🔄 Producer-Consumer

One producer, multiple consumers. Each message processed once. Perfect for task distribution.

📢 Pub-Sub

One publisher, multiple subscribers. Each gets copy. Perfect for event broadcasting.

✅ Delivery Guarantees

At-least-once most common. Requires idempotent consumers. Exactly-once is hardest but most reliable.

📊 Ordering Matters

Message ordering critical for state changes. Per-partition ordering balances order and parallelism.

---

Next Steps

• Learn Kafka Deep Dive - distributed streaming platform
• Understand RabbitMQ - traditional message broker
• Master Event-Driven Architecture - event sourcing and CQRS