Message Queues Fundamentals

Decoupling systems with asynchronous messaging

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.

Key Benefits

Decoupling - Services don’t know about each other
Asynchronous - Non-blocking communication
Reliability - Messages persisted, can retry
Scalability - Multiple consumers, load distribution
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

1
import queue
2
import threading
3
import time
4
from typing import Optional
5

6
class MessageQueue:
7
    """Simple message queue implementation"""
8

9
    def __init__(self, maxsize: int = 1000):
10
        self.queue = queue.Queue(maxsize=maxsize)
11
        self.consumers = []
12
        self.running = False
13

14
    def produce(self, message: dict):
15
        """Producer: Add message to queue"""
16
        try:
17
            self.queue.put_nowait(message)
18
            print(f"Produced: {message}")
19
        except queue.Full:
20
            print("Queue full! Message rejected.")
21

22
    def consume(self, handler):
23
        """Consumer: Process messages from queue"""
24
        while self.running:
25
            try:
26
                message = self.queue.get(timeout=1)
27
                # Process message
28
                handler(message)
29
                # Acknowledge
30
                self.queue.task_done()
31
            except queue.Empty:
32
                continue
33

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

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

44
    def start(self):
45
        """Start queue processing"""
46
        self.running = True
47

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

53
# Usage
54
queue = MessageQueue(maxsize=100)
55
queue.start()
56

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

64
# Start multiple consumers
65
queue.start_consumer(process_order, consumer_id=1)
66
queue.start_consumer(process_order, consumer_id=2)
67

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

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

1
import java.util.concurrent.BlockingQueue;
2
import java.util.concurrent.LinkedBlockingQueue;
3
import java.util.concurrent.ExecutorService;
4
import java.util.concurrent.Executors;
5
import java.util.function.Consumer;
6

7
public class MessageQueue {
8
    private final BlockingQueue<Message> queue;
9
    private final ExecutorService executor;
10
    private volatile boolean running = false;
11

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

17
    public void produce(Message message) {
18
        // Producer: Add message to queue
19
        if (queue.offer(message)) {
20
            System.out.println("Produced: " + message);
21
        } else {
22
            System.out.println("Queue full! Message rejected.");
23
        }
24
    }
25

26
    public void consume(Consumer<Message> handler) {
27
        // Consumer: Process messages from queue
28
        while (running) {
29
            try {
30
                Message message = queue.take();  // Blocks until message available
31
                handler.accept(message);
32
            } catch (InterruptedException e) {
33
                Thread.currentThread().interrupt();
34
                break;
35
            }
36
        }
37
    }
38

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

47
    public void start() {
48
        running = true;
49
    }
50

51
    public void stop() {
52
        running = false;
53
        executor.shutdown();
54
    }
55
}
56

57
// Usage
58
MessageQueue queue = new MessageQueue(1000);
59
queue.start();
60

61
// Message handler
62
Consumer<Message> processOrder = message -> {
63
    System.out.println("Processing order: " + message.getOrderId());
64
    // Process order...
65
    try {
66
        Thread.sleep(1000);  // Simulate work
67
    } catch (InterruptedException e) {
68
        Thread.currentThread().interrupt();
69
    }
70
    System.out.println("Order " + message.getOrderId() + " processed");
71
};
72

73
// Start multiple consumers
74
queue.startConsumer(processOrder, 1);
75
queue.startConsumer(processOrder, 2);
76

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

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

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
Java

1
from typing import List, Callable, Dict, Any
2
from threading import Lock
3
import threading
4

5
class Topic:
6
    """Pub-sub topic"""
7

8
    def __init__(self, name: str):
9
        self.name = name
10
        self.subscribers: List[Callable] = []
11
        self.lock = Lock()
12

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

19
    def unsubscribe(self, handler: Callable):
20
        """Unsubscribe from topic"""
21
        with self.lock:
22
            if handler in self.subscribers:
23
                self.subscribers.remove(handler)
24

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

30
        # Notify all subscribers (asynchronously)
31
        for subscriber in subscribers:
32
            try:
33
                # Run in separate thread to avoid blocking
34
                threading.Thread(
35
                    target=subscriber,
36
                    args=(message,),
37
                    daemon=True
38
                ).start()
39
            except Exception as e:
40
                print(f"Error notifying subscriber: {e}")
41

42
class PubSubBroker:
43
    """Pub-sub message broker"""
44

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

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

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

61
    def subscribe(self, topic_name: str, handler: Callable):
62
        """Subscribe to topic"""
63
        topic = self.get_topic(topic_name)
64
        topic.subscribe(handler)
65

66
# Usage
67
broker = PubSubBroker()
68

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

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

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

79
# Subscribe to 'user.created' topic
80
broker.subscribe('user.created', email_handler)
81
broker.subscribe('user.created', sms_handler)
82
broker.subscribe('user.created', analytics_handler)
83

84
# Publisher publishes event
85
broker.publish('user.created', {
86
    'event': 'user.created',
87
    'user_id': 123,
88
    'email': '[email protected]'
89
})

1
import java.util.*;
2
import java.util.concurrent.CopyOnWriteArrayList;
3
import java.util.function.Consumer;
4

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

9
    public Topic(String name) {
10
        this.name = name;
11
    }
12

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

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

22
    public void publish(Message message) {
23
        // Notify all subscribers
24
        subscribers.forEach(subscriber -> {
25
            try {
26
                subscriber.accept(message);
27
            } catch (Exception e) {
28
                System.err.println("Error notifying subscriber: " + e.getMessage());
29
            }
30
        });
31
    }
32
}
33

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

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

41
    public void publish(String topicName, Message message) {
42
        Topic topic = getTopic(topicName);
43
        topic.publish(message);
44
    }
45

46
    public void subscribe(String topicName, Consumer<Message> handler) {
47
        Topic topic = getTopic(topicName);
48
        topic.subscribe(handler);
49
    }
50
}
51

52
// Usage
53
PubSubBroker broker = new PubSubBroker();
54

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

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

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

65
// Subscribe to 'user.created' topic
66
broker.subscribe("user.created", emailHandler);
67
broker.subscribe("user.created", smsHandler);
68
broker.subscribe("user.created", analyticsHandler);
69

70
// Publisher publishes event
71
broker.publish("user.created", new Message("user.created", 123, "[email protected]"));

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

1
Message 1: Balance = 100
2
Message 2: Balance = 150 (add 50)
3
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

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
Java

1
from abc import ABC, abstractmethod
2
from typing import Dict, Any, Optional
3
import json
4

5
class MessageHandler(ABC):
6
    """Base message handler interface"""
7

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

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

21
class OrderProcessor(MessageHandler):
22
    """Process order messages"""
23

24
    def __init__(self, order_service):
25
        self.order_service = order_service
26
        self.processed_ids = set()  # For idempotency
27

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

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

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

39
        try:
40
            # Process order
41
            self.order_service.process_order(order_id, message)
42

43
            # Mark as processed
44
            self.processed_ids.add(order_id)
45

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

51
class MessageConsumer:
52
    """Consumer that routes messages to handlers"""
53

54
    def __init__(self, queue, handlers: List[MessageHandler]):
55
        self.queue = queue
56
        self.handlers = handlers
57

58
    def consume(self):
59
        """Consume messages from queue"""
60
        while True:
61
            try:
62
                message_data = self.queue.get(timeout=1)
63
                message = json.loads(message_data)
64

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

68
                if handler:
69
                    # Process message
70
                    success = handler.handle(message)
71

72
                    if success:
73
                        # Acknowledge message
74
                        self.queue.task_done()
75
                    else:
76
                        # Return to queue for retry
77
                        self.queue.put(message_data)
78
                else:
79
                    print(f"No handler for message type: {message.get('type')}")
80
                    self.queue.task_done()
81

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

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

1
import java.util.*;
2

3
interface MessageHandler {
4
    boolean handle(Message message);
5
    boolean canHandle(String messageType);
6
}
7

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

12
    public OrderProcessor(OrderService orderService) {
13
        this.orderService = orderService;
14
    }
15

16
    @Override
17
    public boolean canHandle(String messageType) {
18
        return "order.created".equals(messageType);
19
    }
20

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

25
        // Idempotency check
26
        synchronized (processedIds) {
27
            if (processedIds.contains(orderId)) {
28
                System.out.println("Order " + orderId + " already processed. Skipping.");
29
                return true;  // Already processed
30
            }
31
        }
32

33
        try {
34
            // Process order
35
            orderService.processOrder(orderId, message);
36

37
            // Mark as processed
38
            synchronized (processedIds) {
39
                processedIds.add(orderId);
40
            }
41

42
            return true;
43
        } catch (Exception e) {
44
            System.err.println("Error processing order " + orderId + ": " + e.getMessage());
45
            return false;  // Return false to trigger retry
46
        }
47
    }
48
}
49

50
class MessageConsumer {
51
    private final BlockingQueue<String> queue;
52
    private final List<MessageHandler> handlers;
53

54
    public MessageConsumer(BlockingQueue<String> queue, List<MessageHandler> handlers) {
55
        this.queue = queue;
56
        this.handlers = handlers;
57
    }
58

59
    public void consume() {
60
        while (true) {
61
            try {
62
                String messageData = queue.take();
63
                Message message = parseMessage(messageData);
64

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

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

72
                    if (success) {
73
                        // Message processed successfully
74
                        // (Acknowledgment handled by queue)
75
                    } else {
76
                        // Return to queue for retry
77
                        queue.put(messageData);
78
                    }
79
                } else {
80
                    System.err.println("No handler for message type: " + message.getType());
81
                }
82
            } catch (InterruptedException e) {
83
                Thread.currentThread().interrupt();
84
                break;
85
            } catch (Exception e) {
86
                System.err.println("Error consuming message: " + e.getMessage());
87
            }
88
        }
89
    }
90

91
    private MessageHandler findHandler(String messageType) {
92
        return handlers.stream()
93
            .filter(h -> h.canHandle(messageType))
94
            .findFirst()
95
            .orElse(null);
96
    }
97
}

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

Message Queues Fundamentals

The Synchronous Problem

What is a Message Queue?

Key Benefits

Pattern 1: Producer-Consumer

Producer-Consumer Implementation

Pattern 2: Pub-Sub (Publish-Subscribe)

Pub-Sub Implementation

Message Delivery Guarantees

At-Most-Once (Fire and Forget)

At-Least-Once (Guaranteed Delivery)

Exactly-Once (Hardest)

Message Ordering

Why Ordering Matters

Ordering Strategies

When to Use Message Queues

Use Queues When:

Don’t Use Queues When:

LLD Connection: Implementing Message Handlers

Message Handler Design

Key Takeaways

Next Steps