Message Queue Components Explained — Producers, Brokers, Consumers and Beyond

import pika
import json
import uuid
from datetime import datetime

# Connect to RabbitMQ running locally.
# In production this would point at your cluster address.
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost')
)
channel = connection.channel()

# Declare the queue before publishing.
# durable=True means the queue survives a broker restart.
channel.queue_declare(queue='order_processing', durable=True)

def publish_order(customer_id: str, items: list, total_usd: float):
    """Build a structured order event and push it to the queue."""

    order_event = {
        "event_id": str(uuid.uuid4()),        # Unique ID — critical for deduplication
        "event_type": "order.placed",
        "customer_id": customer_id,
        "items": items,
        "total_usd": total_usd,
        "created_at": datetime.utcnow().isoformat()
    }

    channel.basic_publish(
        exchange='',                          # Default exchange routes by queue name
        routing_key='order_processing',       # Must match the declared queue name
        body=json.dumps(order_event),
        properties=pika.BasicProperties(
            delivery_mode=2,                  # 2 = persistent (written to disk by broker)
            content_type='application/json'
        )
    )

    print(f"[Producer] Order {order_event['event_id']} published.")
    return order_event['event_id']

# Simulate two customers placing orders
publish_order(
    customer_id='cust-881',
    items=[{'sku': 'BOOT-42', 'qty': 1}, {'sku': 'LACE-BLK', 'qty': 2}],
    total_usd=89.99
)

publish_order(
    customer_id='cust-334',
    items=[{'sku': 'HAT-ML', 'qty': 1}],
    total_usd=29.95
)

connection.close()
print("[Producer] Connection closed. Producer is done — it does NOT wait for consumers.")

# This demo uses Python's built-in queue module to simulate
# broker behaviour locally — no external dependency needed.
# It illustrates queue vs topic routing at the broker level.

import queue
import threading
import time

# ── QUEUE SEMANTICS (point-to-point) ────────────────────────────────────────
# Broker holds messages; only ONE consumer gets each message.
payment_queue = queue.Queue(maxsize=100)  # maxsize adds backpressure

def simulate_queue_broker():
    print("\n=== QUEUE (Point-to-Point) Demo ===")

    # Two workers compete for the same messages
    messages = ["payment:order-1", "payment:order-2", "payment:order-3"]
    for msg in messages:
        payment_queue.put(msg)  # Producer pushes message to broker

    def payment_worker(worker_id: int):
        while not payment_queue.empty():
            try:
                # get_nowait raises Empty if nothing left — simulates
                # the broker handing off and removing the message
                msg = payment_queue.get_nowait()
                print(f"  Worker-{worker_id} processed: {msg}")
                payment_queue.task_done()  # Acknowledge to broker
                time.sleep(0.05)
            except queue.Empty:
                break  # Another worker grabbed it first — that's correct behaviour

    workers = [
        threading.Thread(target=payment_worker, args=(1,)),
        threading.Thread(target=payment_worker, args=(2,))
    ]
    for w in workers: w.start()
    for w in workers: w.join()
    print("  Note: each message was processed by exactly ONE worker.")


# ── TOPIC SEMANTICS (publish-subscribe) ─────────────────────────────────────
# One message fans out to EVERY subscriber.
subscriber_queues = {
    "inventory_service": queue.Queue(),
    "email_service": queue.Queue(),
    "analytics_service": queue.Queue()
}

def publish_to_topic(event: dict):
    """Broker fans out the event to every subscriber's queue."""
    for service_name, subscriber_queue in subscriber_queues.items():
        subscriber_queue.put(event)  # Every subscriber gets a copy
        print(f"  Broker delivered '{event['type']}' to {service_name}")

def simulate_topic_broker():
    print("\n=== TOPIC (Pub-Sub) Demo ===")
    order_placed_event = {"type": "order.placed", "order_id": "ord-555", "total": 49.99}
    publish_to_topic(order_placed_event)
    print(f"  All {len(subscriber_queues)} subscribers received the same event.")


simulate_queue_broker()
simulate_topic_broker()

import pika
import json
import time
import random

# Connect to the same RabbitMQ instance the producer used
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost')
)
channel = connection.channel()

# Declare the queue again — safe to call multiple times, it's idempotent
channel.queue_declare(queue='order_processing', durable=True)

# prefetch_count=1 tells the broker: "don't send me another message until
# I've acked this one". Without this, a slow consumer piles up messages
# in memory and defeats the purpose of distributing load.
channel.basic_qos(prefetch_count=1)

def process_order(channel, method, properties, body):
    """
    This function is called by pika every time a new message arrives.
    'method' contains delivery metadata (delivery_tag for acking).
    """
    order_event = json.loads(body)

    order_id = order_event.get('event_id', 'unknown')
    customer_id = order_event.get('customer_id', 'unknown')
    total = order_event.get('total_usd', 0.0)

    print(f"[Consumer] Received order {order_id} for customer {customer_id} (${total})")

    # Simulate real work — database write, fraud check, warehouse API call
    processing_time = random.uniform(0.5, 1.5)
    time.sleep(processing_time)

    # Simulate an occasional processing failure (10% chance)
    if random.random() < 0.10:
        print(f"[Consumer] FAILED to process {order_id} — sending NACK for requeue")
        # nack with requeue=True: broker puts the message back in the queue
        # for another consumer to retry. requeue=False sends to dead-letter queue.
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        return

    # SUCCESS — tell the broker it can delete this message
    channel.basic_ack(delivery_tag=method.delivery_tag)
    print(f"[Consumer] Order {order_id} processed in {processing_time:.2f}s — ACK sent")

# Register our callback and start listening
channel.basic_consume(
    queue='order_processing',
    on_message_callback=process_order
)

print("[Consumer] Waiting for orders. Press CTRL+C to stop.")
channel.start_consuming()  # Blocking loop — consumer runs until interrupted

import pika
import json
import time

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

# ── Step 1: Declare the Dead-Letter Queue first ──────────────────────────────
# The DLQ is just a normal queue — no special broker type needed.
channel.queue_declare(queue='order_processing_dlq', durable=True)

# ── Step 2: Declare a dead-letter exchange that routes to the DLQ ────────────
channel.exchange_declare(
    exchange='order_dlx',      # Dead-letter exchange name
    exchange_type='direct',
    durable=True
)
channel.queue_bind(
    queue='order_processing_dlq',
    exchange='order_dlx',
    routing_key='order_processing'  # Must match the main queue name
)

# ── Step 3: Declare the MAIN queue WITH dead-letter configuration ────────────
# x-dead-letter-exchange: where to send failed messages
# x-message-ttl: message expires after 60000ms if not consumed (optional)
# x-max-length: if queue exceeds 10000 messages, oldest are DLQ'd (backpressure)
channel.queue_declare(
    queue='order_processing',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'order_dlx',
        'x-dead-letter-routing-key': 'order_processing',
        'x-message-ttl': 60000,
        'x-max-length': 10000
    }
)

print("[Setup] Main queue and DLQ configured successfully.")

# ── Step 4: Consumer that rejects a bad message into the DLQ ─────────────────
def process_with_dlq(channel, method, properties, body):
    try:
        order = json.loads(body)

        # Validate required fields — a real poison message scenario
        if 'customer_id' not in order or 'total_usd' not in order:
            raise ValueError(f"Malformed order event: missing required fields")

        print(f"[Consumer] Processed order {order.get('event_id')}")
        channel.basic_ack(delivery_tag=method.delivery_tag)

    except (json.JSONDecodeError, ValueError) as error:
        print(f"[Consumer] Poison message detected: {error}")
        # requeue=False is KEY — tells broker 'don't retry this,
        # route it to the dead-letter exchange instead'
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        print("[Consumer] Message sent to DLQ for manual inspection.")

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='order_processing', on_message_callback=process_with_dlq)

# Publish a deliberately malformed message to trigger DLQ routing
channel.basic_publish(
    exchange='',
    routing_key='order_processing',
    body=json.dumps({"event_id": "bad-001", "broken": True}),  # Missing customer_id and total_usd
    properties=pika.BasicProperties(delivery_mode=2)
)

print("[Setup] Consuming one message then exiting...")
connection.process_data_events(time_limit=2)  # Process one cycle
connection.close()