Message Queues & Pub/Sub

Metadata

1. Overview

What problem does it solve?

When Service A calls Service B over a synchronous HTTP REST API, they are tightly coupled.

1. Service A has to wait for Service B to finish.
2. If Service B is down, Service A's request fails.
3. If Service A gets a massive spike in traffic, Service B gets crushed instantly.

A Message Queue is an asynchronous communication tool that sits between services. Service A throws a message into the queue and immediately returns to the user. Service B pulls messages from the queue at its own pace.

What breaks without it?

• Spiky Traffic: A video processing app allows users to upload videos. Processing takes 5 minutes. If 1,000 users upload at once, synchronous HTTP will freeze the whole system and crash the servers.
• Microservice Resiliency: Every time a downstream service goes down for 5 minutes, you lose 5 minutes of user data because the HTTP requests failed.

2. Motivation

As monolithic applications broke apart into Microservices, the network between them became a massive point of failure. Early Service-Oriented Architectures (SOA) tried to solve this with Enterprise Service Buses (ESBs), which were bloated and complex.

Modern distributed systems shifted to "smart endpoints and dumb pipes" (using simple Message Brokers like RabbitMQ or ActiveMQ). Later, as Big Data exploded, LinkedIn invented Kafka — an append-only distributed log that revolutionized high-throughput Pub/Sub messaging.

3. Real-World Usage

4. Prerequisites

5. Visual Explanation

Pattern 1: Point-to-Point (Work Queue)

One producer, multiple consumers. A message is processed by exactly ONE consumer. Use Case: Distributing heavy background jobs (like video encoding).

%%{init: {'theme': 'base', 'themeVariables': { 'primaryColor': '#ffcc00', 'edgeLabelBackground':'#ffffff'}}}%%
graph LR
    classDef svc fill:#f9f9f9,stroke:#333,stroke-width:2px;
    classDef queue fill:#d4edda,stroke:#28a745,stroke-width:2px;

    P[Producer]:::svc --> |"Message A, B, C"| Q[(Message Queue)]:::queue
    
    Q --> |"Message A"| C1[Consumer 1]:::svc
    Q --> |"Message B"| C2[Consumer 2]:::svc
    Q --> |"Message C"| C1

Pattern 2: Publish / Subscribe (Pub/Sub)

One producer, multiple consumers. A message is broadcast to ALL consumers. Use Case: An order is placed. The Email Service needs to know, the Shipping Service needs to know, and the Analytics Service needs to know.

graph LR
    classDef svc fill:#f9f9f9,stroke:#333,stroke-width:2px;
    classDef topic fill:#cce5ff,stroke:#007bff,stroke-width:2px;

    P[Producer]:::svc --> |"Order #123"| T{{Topic: 'Orders'}}:::topic
    
    T --> |"Order #123"| C1[Email Service]:::svc
    T --> |"Order #123"| C2[Shipping Service]:::svc
    T --> |"Order #123"| C3[Analytics Service]:::svc

6. Internal Working

Message Acknowledgment (ACK)

How does the queue know a message was successfully processed? If the broker deletes the message the second it sends it to the consumer, and the consumer crashes before finishing the work, the data is lost.

1. Consumer pulls the message. (Message is "hidden" but not deleted).
2. Consumer does the work (e.g., resizes the image).
3. Consumer sends an ACK back to the broker.
4. Broker safely deletes the message.

If the consumer crashes, the broker waits for a timeout. If no ACK arrives, the message becomes visible again for another consumer to pick up. This guarantees At-Least-Once Delivery.

Delivery Guarantees

1. At-Most-Once (Fire and Forget): Fastest. Data loss is acceptable (e.g., IoT sensor telemetry).
2. At-Least-Once: Standard. Message is guaranteed to arrive, but might arrive twice if the ACK is lost.
3. Exactly-Once: Extremely hard. Usually requires idempotency on the consumer side.

7. Implementation

Why Python? Python's built-in queue module and threading allow us to build a fully functional in-memory Message Broker demonstrating both Point-to-Point and Pub/Sub architectures in under 100 lines.

"""
024 - Message Queues & Pub/Sub
A simulated in-memory message broker supporting both Work Queues
and Pub/Sub Topics using threads.
"""
import time
import threading
from queue import Queue

# ── 1. Point-to-Point (Work Queue) ──

class WorkQueue:
    def __init__(self):
        self.q = Queue()

    def produce(self, message):
        print(f"[Producer] Added to queue: '{message}'")
        self.q.put(message)

    def consume(self, worker_id):
        while True:
            # Block until a message is available
            message = self.q.get()
            if message is None: # Poison pill to stop
                break
                
            print(f"[Worker-{worker_id}] Processing: '{message}'")
            time.sleep(0.5) # Simulate hard work
            
            # Acknowledge completion
            self.q.task_done()
            print(f"[Worker-{worker_id}] Finished and ACKed: '{message}'")


# ── 2. Publish / Subscribe (Topic) ──

class PubSubBroker:
    def __init__(self):
        # Maps Topic Name -> List of Subscriber Queues
        self.topics = {}
        self.lock = threading.Lock()

    def subscribe(self, topic_name, subscriber_id):
        with self.lock:
            if topic_name not in self.topics:
                self.topics[topic_name] = []
            
            # Give the subscriber their own personal queue
            sub_queue = Queue()
            self.topics[topic_name].append((subscriber_id, sub_queue))
            return sub_queue

    def publish(self, topic_name, message):
        with self.lock:
            if topic_name not in self.topics:
                return
            
            print(f"[Publisher] Broadcasting to '{topic_name}': {message}")
            # Duplicate the message to EVERY subscriber's personal queue
            for sub_id, sub_queue in self.topics[topic_name]:
                sub_queue.put(message)


def subscriber_worker(sub_id, queue):
    while True:
        msg = queue.get()
        if msg is None:
            break
        print(f"[Subscriber-{sub_id}] Received broadcast: {msg}")


# ── Test Harness ──

if __name__ == "__main__":
    print("=== PART 1: Work Queue (Point-to-Point) ===")
    wq = WorkQueue()

    # Start 2 competing consumers
    t1 = threading.Thread(target=wq.consume, args=(1,))
    t2 = threading.Thread(target=wq.consume, args=(2,))
    t1.start()
    t2.start()

    # Produce 4 tasks
    for i in range(1, 5):
        wq.produce(f"Task {i}")

    # Wait for work queue to empty
    wq.q.join()
    
    # Send poison pills to shut down threads
    wq.q.put(None)
    wq.q.put(None)
    t1.join()
    t2.join()

    print("\n=== PART 2: Pub/Sub ===")
    broker = PubSubBroker()

    # Create 3 subscribers
    q_email = broker.subscribe("orders", "EmailService")
    q_shipping = broker.subscribe("orders", "ShippingService")
    q_analytics = broker.subscribe("orders", "AnalyticsService")

    # Start subscriber threads
    threads = []
    for sub_id, q in [("Email", q_email), ("Shipping", q_shipping), ("Analytics", q_analytics)]:
        t = threading.Thread(target=subscriber_worker, args=(sub_id, q))
        t.start()
        threads.append(t)

    # Publish ONE message
    time.sleep(0.1)
    broker.publish("orders", '{"order_id": 99, "amount": 50}')
    
    time.sleep(0.5)
    # Cleanup
    for q in [q_email, q_shipping, q_analytics]:
        q.put(None)
    for t in threads:
        t.join()

Sample Output

=== PART 1: Work Queue (Point-to-Point) ===
[Producer] Added to queue: 'Task 1'
[Worker-1] Processing: 'Task 1'
[Producer] Added to queue: 'Task 2'
[Worker-2] Processing: 'Task 2'
[Producer] Added to queue: 'Task 3'
[Producer] Added to queue: 'Task 4'
[Worker-1] Finished and ACKed: 'Task 1'
[Worker-1] Processing: 'Task 3'
[Worker-2] Finished and ACKed: 'Task 2'
[Worker-2] Processing: 'Task 4'
[Worker-1] Finished and ACKed: 'Task 3'
[Worker-2] Finished and ACKed: 'Task 4'

=== PART 2: Pub/Sub ===
[Publisher] Broadcasting to 'orders': {"order_id": 99, "amount": 50}
[Subscriber-Email] Received broadcast: {"order_id": 99, "amount": 50}
[Subscriber-Shipping] Received broadcast: {"order_id": 99, "amount": 50}
[Subscriber-Analytics] Received broadcast: {"order_id": 99, "amount": 50}

Notice how in Part 1, the tasks were distributed evenly among workers. In Part 2, the single message was duplicated to every subscriber.

8. Complexity

9. Trade-offs

10. Production Evolution

11. Common Bugs

12. Interview Questions

1. What is the difference between RabbitMQ (Message Broker) and Kafka (Event Streaming)? Hint: RabbitMQ pushes messages, deletes them on ACK, and is great for task queues. Kafka is an immutable log on disk; consumers pull messages, track their own offsets, and can replay history.

2. How do Message Queues handle sudden spikes in traffic? Hint: Load Leveling. The producers dump messages into the queue quickly. The queue holds them safely while the consumers chew through them at their maximum safe capacity.

3. What is a Dead Letter Queue (DLQ)? Hint: A parking lot for poisonous messages that repeatedly crash consumers, allowing the main queue to continue flowing.

4. Why are Message Queues considered "Eventually Consistent"? Hint: The producer gets a "Success" from the queue, but the actual database update might not happen until the consumer finishes processing it seconds or minutes later.

13. Used By (Downstream Blocks)

• 025 Event-Driven Architecture — Entire architectures built around Pub/Sub.
• 026 Distributed Transactions (Sagas) — Uses message queues to coordinate complex transactions across multiple microservices.

14. Used In (Case Studies)

15. Related Blocks

16. Try It Yourself

Exercise 1: Dead Letter Queue (DLQ)

Modify the WorkQueue implementation. If a message contains the string "ERROR", the consumer should fail (do not ACK it). Add retry logic to the queue: if a message fails 3 times, route it to a separate self.dlq = Queue() and print a warning.

Exercise 2: Fan-out Exchange

In RabbitMQ, a "Fan-out Exchange" routes one message to multiple different Work Queues. Combine the two models: Create a Pub/Sub topic where each subscriber is actually a WorkQueue with multiple competing threads.

Website Metadata