RabbitMQ #

RabbitMQ is an open-source message broker implementing the AMQP (Advanced Message Queuing Protocol) — a standard protocol for inter-application message exchange that guarantees reliable delivery, routing, and message queuing. Unlike Kafka, designed as a distributed log, RabbitMQ is a classic broker with a push model: messages are delivered to consumers as soon as they’re available and removed from the queue after being acknowledged. RabbitMQ’s strength lies in routing flexibility through Exchanges with various types, message TTL support, Dead Letter Exchanges, and priority queues — making it ideal for task queues, job scheduling, and microservice communication that needs complex routing.

RabbitMQ Basic Concepts #

flowchart LR
    subgraph Producer
        A["App A"]
    end

    subgraph Broker["RabbitMQ Broker"]
        E["Exchange"]
        
        subgraph Queues["Queue(s)"]
            QA["Queue A"]
            QB["Queue B"]
            QC["Queue C"]
        end
    end

    subgraph Consumers["Consumer"]
        W1["Worker 1"]
        W2["Worker 2"]
        W3["Worker 3"]
    end

    A --> |publish| E
    E --> |binding| QA
    E --> |binding| QB
    E --> |binding| QC
    QA --> W1
    QB --> W2
    QC --> W3

Exchange types:

direct: routing based on exact routing-key match (Producer → "order.created" → Queue bound with key "order.created")
fanout: broadcasts to ALL bound queues, the routing key is ignored (Producer → Exchange → Queue A + Queue B + Queue C)
topic: routing based on pattern matching with wildcards (* and #)
- * = one word, # = zero or more words
- "order.*" → matches "order.created", "order.paid"
- "order.#" → matches "order.created", "order.paid.success"
headers: routing based on message headers, not the routing key

Installation #

pip install pika

To run RabbitMQ locally using Docker:

docker run -d \
  --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=secret \
  rabbitmq:3-management
# Management UI available at http://localhost:15672

Creating a Connection #

import pika
import os
from urllib.parse import quote

# ANTI-PATTERN: hardcoded connection without authentication
connection = pika.BlockingConnection(
    pika.ConnectionParameters("localhost")  # ✗ -- no user/pass, not flexible
)

# CORRECT: use environment variables and URLParameters
def get_connection() -> pika.BlockingConnection:
    amqp_url = os.getenv(
        "RABBITMQ_URL",
        "amqp://admin:***@localhost:5672/%2F"
        # Format: amqp://user:***@host:port/vhost
        # %2F is the URL-encoded "/" for the default vhost
    )
    params = pika.URLParameters(amqp_url)
    params.heartbeat            = 60     # send a heartbeat every 60 seconds
    params.blocked_connection_timeout = 300

    return pika.BlockingConnection(params)

# Test the connection
conn    = get_connection()
channel = conn.channel()
print("RabbitMQ connection successful.")
conn.close()

Setting Up Exchanges, Queues, and Bindings #

Always declare the exchange and queue on both sides (producer and consumer) with the same parameters — RabbitMQ is idempotent for identical declarations.

import pika
import json

def setup_infrastruktur(channel: pika.channel.Channel) -> None:
    """
    Declare the exchange, queue, and bindings.
    Safe to call repeatedly -- idempotent.
    """

    # Main exchange for order events
    channel.exchange_declare(
        exchange="order.events",
        exchange_type="topic",
        durable=True       # survives a RabbitMQ restart
    )

    # Dead Letter Exchange -- for messages that failed processing
    channel.exchange_declare(
        exchange="order.dlx",
        exchange_type="direct",
        durable=True
    )

    # Queue for the payment service
    channel.queue_declare(
        queue="payment.queue",
        durable=True,      # the queue survives a restart
        arguments={
            "x-dead-letter-exchange":    "order.dlx",        # send to the DLX on failure
            "x-dead-letter-routing-key": "payment.dead",
            "x-message-ttl":             86400000,           # TTL of 24 hours (ms)
            "x-max-length":              10000               # max 10,000 messages
        }
    )

    # Queue for the notification service
    channel.queue_declare(
        queue="notification.queue",
        durable=True,
        arguments={
            "x-dead-letter-exchange":    "order.dlx",
            "x-dead-letter-routing-key": "notification.dead",
        }
    )

    # Dead Letter Queue -- holds failed messages
    channel.queue_declare(queue="order.dead.queue", durable=True)

    # Binding: exchange → queue with a routing key
    channel.queue_bind(
        exchange="order.events",
        queue="payment.queue",
        routing_key="order.created"     # only receive "order.created" events
    )
    channel.queue_bind(
        exchange="order.events",
        queue="payment.queue",
        routing_key="order.updated"
    )
    channel.queue_bind(
        exchange="order.events",
        queue="notification.queue",
        routing_key="order.*"           # receive all "order.{anything}"
    )

    # DLX bindings
    channel.queue_bind(
        exchange="order.dlx",
        queue="order.dead.queue",
        routing_key="payment.dead"
    )
    channel.queue_bind(
        exchange="order.dlx",
        queue="order.dead.queue",
        routing_key="notification.dead"
    )

    print("Exchange, queue, and bindings declared successfully.")

Producer — Sending Messages #

import pika
import json
from datetime import datetime, timezone

def buat_producer() -> tuple[pika.BlockingConnection, pika.channel.Channel]:
    conn    = get_connection()
    channel = conn.channel()
    setup_infrastruktur(channel)
    return conn, channel

def kirim_event(
    channel:     pika.channel.Channel,
    exchange:    str,
    routing_key: str,
    payload:     dict
) -> None:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")

    channel.basic_publish(
        exchange=exchange,
        routing_key=routing_key,
        body=body,
        properties=pika.BasicProperties(
            content_type  = "application/json",
            delivery_mode = pika.DeliveryMode.Persistent,  # ✓ message saved to disk
            message_id    = payload.get("event_id"),
            timestamp     = int(datetime.now(timezone.utc).timestamp()),
            headers       = {
                "source":  "order-service",
                "version": "1.0"
            }
        )
    )

# Usage
conn, channel = buat_producer()

try:
    # Send an order.created event
    order_event = {
        "event_id":  "evt-001",
        "event":     "order.created",
        "order_id":  1001,
        "user_id":   42,
        "total":     18500000,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }
    kirim_event(channel, "order.events", "order.created", order_event)
    print(f"Event sent: {order_event['event']} (order #{order_event['order_id']})")

    # Send an order.updated event
    update_event = {
        "event_id": "evt-002",
        "event":    "order.updated",
        "order_id": 1001,
        "status":   "paid",
        "timestamp": datetime.now(timezone.utc).isoformat()
    }
    kirim_event(channel, "order.events", "order.updated", update_event)

finally:
    conn.close()

Always use delivery_mode=Persistent for important messages. Without it (delivery_mode=Transient), messages are only stored in memory and will be lost if RabbitMQ restarts or crashes. Persistent messages combined with durable=True on the queue are the pair that guarantees message durability.

Consumer — Receiving Messages #

import pika
import json
import signal

def proses_pembayaran(order: dict) -> None:
    """Payment processing business logic."""
    print(f"Processing payment for order #{order['order_id']} — Rp{order['total']:,.0f}")
    # If an exception is raised here, the message will be nacked and go to the DLX

def jalankan_payment_consumer() -> None:
    conn    = get_connection()
    channel = conn.channel()
    setup_infrastruktur(channel)

    # QoS: process only 1 message per consumer at a time
    # Ensures the load is distributed evenly among workers
    channel.basic_qos(prefetch_count=1)

    def callback(ch, method, properties, body):
        try:
            payload = json.loads(body.decode("utf-8"))
            print(f"Received: {properties.message_id} — routing: {method.routing_key}")

            proses_pembayaran(payload)

            # ACK -- tell RabbitMQ the message is processed, remove it from the queue
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"✓ Message ACKed: {properties.message_id}")

        except json.JSONDecodeError as e:
            print(f"✗ Invalid message (JSON error): {e}")
            # Reject without requeue -- straight to the DLX
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

        except Exception as e:
            print(f"✗ Error processing message: {e}")
            # ANTI-PATTERN: basic_ack on error
            # ch.basic_ack(delivery_tag=method.delivery_tag)  # ✗ -- message lost!

            # CORRECT: nack with requeue=False → goes to the Dead Letter Exchange
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # ✓

    channel.basic_consume(
        queue="payment.queue",
        on_message_callback=callback,
        auto_ack=False   # MUST be False -- manual commit after processing finishes
    )

    # Graceful shutdown
    def handle_shutdown(signum, frame):
        print("Shutdown signal received, stopping consumer...")
        channel.stop_consuming()

    signal.signal(signal.SIGINT,  handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    print("Payment consumer active, waiting for messages...")
    try:
        channel.start_consuming()
    finally:
        conn.close()
        print("Consumer shut down cleanly.")

jalankan_payment_consumer()

Exchange Types in Practice #

Direct Exchange — Precise Routing #

# Setup
channel.exchange_declare(exchange="notif.direct", exchange_type="direct", durable=True)
channel.queue_declare(queue="email.queue",   durable=True)
channel.queue_declare(queue="sms.queue",     durable=True)
channel.queue_declare(queue="push.queue",    durable=True)

channel.queue_bind(exchange="notif.direct", queue="email.queue", routing_key="email")
channel.queue_bind(exchange="notif.direct", queue="sms.queue",   routing_key="sms")
channel.queue_bind(exchange="notif.direct", queue="push.queue",  routing_key="push")

# Send only to the email queue
channel.basic_publish(
    exchange="notif.direct",
    routing_key="email",   # only email.queue receives it
    body=json.dumps({"to": "[email protected]", "subject": "Order Confirmed"}).encode()
)

Fanout Exchange — Broadcasting #

# Setup -- the routing key is ignored
channel.exchange_declare(exchange="order.broadcast", exchange_type="fanout", durable=True)
channel.queue_declare(queue="audit.queue",   durable=True)
channel.queue_declare(queue="report.queue",  durable=True)
channel.queue_declare(queue="analytics.queue", durable=True)

# Bind all queues (without a routing key)
for q in ["audit.queue", "report.queue", "analytics.queue"]:
    channel.queue_bind(exchange="order.broadcast", queue=q)

# One publish → three queues receive it simultaneously
channel.basic_publish(
    exchange="order.broadcast",
    routing_key="",   # ignored for fanout
    body=json.dumps({"event": "order.completed", "order_id": 1001}).encode()
)

Topic Exchange — Pattern Matching #

# Setup
channel.exchange_declare(exchange="logs", exchange_type="topic", durable=True)
channel.queue_declare(queue="error.queue",   durable=True)
channel.queue_declare(queue="warning.queue", durable=True)
channel.queue_declare(queue="all.queue",     durable=True)

# * = exactly one word, # = zero or more words
channel.queue_bind(exchange="logs", queue="error.queue",   routing_key="*.error")
channel.queue_bind(exchange="logs", queue="warning.queue", routing_key="*.warning")
channel.queue_bind(exchange="logs", queue="all.queue",     routing_key="#")  # receive everything

# "payment.error"   → error.queue + all.queue
# "auth.warning"    → warning.queue + all.queue
# "order.info"      → all.queue only
channel.basic_publish(exchange="logs", routing_key="payment.error",
                      body=b"Payment gateway timeout")
channel.basic_publish(exchange="logs", routing_key="auth.warning",
                      body=b"Failed login attempt")

Automatic Reconnection #

RabbitMQ connections can drop due to network issues or timeouts. It’s important to have reconnection logic so consumers don’t die silently.

import pika
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def jalankan_consumer_dengan_reconnect(
    queue:       str,
    callback_fn,
    maks_retry:  int = 5,
    jeda_retry:  int = 5
) -> None:
    retry = 0

    while True:
        try:
            conn    = get_connection()
            channel = conn.channel()
            setup_infrastruktur(channel)
            channel.basic_qos(prefetch_count=1)

            channel.basic_consume(
                queue=queue,
                on_message_callback=callback_fn,
                auto_ack=False
            )

            logger.info(f"Consumer active on queue '{queue}'")
            retry = 0   # reset the counter after a successful connect
            channel.start_consuming()

        except pika.exceptions.AMQPConnectionError as e:
            retry += 1
            if retry > maks_retry:
                logger.error(f"Failed to reconnect after {maks_retry} attempts. Stopping.")
                raise

            logger.warning(
                f"Connection lost ({e}). "
                f"Reconnecting in {jeda_retry}s... (attempt {retry}/{maks_retry})"
            )
            time.sleep(jeda_retry)

        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            raise

        finally:
            try:
                if conn and not conn.is_closed:
                    conn.close()
            except Exception:
                pass

Error Handling and Dead Letter Exchanges #

A Dead Letter Exchange (DLX) holds messages that failed processing — whether because they were nacked, their TTL expired, or the queue was full. This prevents messages from disappearing entirely.

def jalankan_dlq_consumer() -> None:
    """Consumer to monitor and reprocess messages in the DLQ."""
    conn    = get_connection()
    channel = conn.channel()

    def callback_dlq(ch, method, properties, body):
        try:
            payload  = json.loads(body.decode("utf-8"))
            headers  = properties.headers or {}
            kematian = headers.get("x-death", [{}])[0]

            print(f"[DLQ] Dead message from queue: {kematian.get('queue')}")
            print(f"[DLQ] Reason: {kematian.get('reason')}")
            print(f"[DLQ] Death count: {kematian.get('count')}")
            print(f"[DLQ] Payload: {payload}")

            # Strategy: log, alert, or manual retry
            # After processing, ACK so it doesn't pile up in the DLQ
            ch.basic_ack(delivery_tag=method.delivery_tag)

        except Exception as e:
            print(f"[DLQ] Error: {e}")
            ch.basic_ack(delivery_tag=method.delivery_tag)  # still ACK in the DLQ

    channel.basic_consume(
        queue="order.dead.queue",
        on_message_callback=callback_dlq,
        auto_ack=False
    )

    print("DLQ consumer active...")
    channel.start_consuming()

Summary #

auto_ack=False is mandatory — always use manual acknowledgment; auto_ack=True acknowledges messages immediately upon receipt, before processing, so messages are lost if the consumer crashes.
delivery_mode=Persistent — use it so messages are saved to disk and survive a RabbitMQ restart; pair it with durable=True on the queue.
basic_ack() only after success — call it only after processing succeeds; use basic_nack(requeue=False) to reject failed messages to the DLX.
prefetch_count=1 — set QoS so each consumer only receives one message at a time; prevents one consumer from being flooded while others sit idle.
Dead Letter Exchange (DLX) — always configure a DLX on production queues so failed messages don’t disappear, but instead enter a monitoring queue.
Pick the right exchange type — direct for precise routing, fanout for broadcasting, topic for routing-key pattern matching.
durable=True for exchanges and queues — so broker configuration survives restarts.
Reconnection logic — implement retries with backoff for consumers so they don’t die silently when the connection drops.
x-message-ttl — use TTL on queues to automatically clean up old unprocessed messages.
Idempotent declarations — declare exchanges, queues, and bindings on both sides (producer and consumer) with identical parameters; RabbitMQ won’t error if they already exist.

← Previous: Kafka Next: Amazon SQS →