RabbitMQ #

RabbitMQ is a traditional message broker implementing the AMQP (Advanced Message Queuing Protocol). Unlike Kafka, which stores logs permanently, RabbitMQ deletes messages after consumption — better suited for task queues, job distribution, and RPC. Its strengths: very flexible routing through the Exchange system, built-in dead letter queues, and easy setup. Go uses the github.com/rabbitmq/amqp091-go library.

Basic RabbitMQ Concepts #

Producer   → sends messages to an Exchange
Exchange   → receives messages and routes them to Queues based on rules
Binding    → the rule connecting an Exchange to a Queue
Queue      → the queue where messages wait to be consumed
Consumer   → reads messages from a Queue

Exchange Types:
  direct  → routing based on an exact routing key match
  fanout  → broadcasts to all bound queues
  topic   → routing keys with wildcards (* and #)
  headers → routing based on message headers

Visualizing the RabbitMQ AMQP Architecture #

Here’s a diagram of how a message flows from the Producer through the Exchange, is directed by Bindings, enters Queues, and is finally processed by a Consumer:

flowchart TD
    P[Producer] -->|"Publish Message + Routing Key"| Ex["Exchange"]
    Ex -->|"Binding Key: 'order.created'"| Q1["Queue 1 (Process Order)"]
    Ex -->|"Binding Key: 'order.deleted'"| Q2["Queue 2 (Notifications)"]

    Q1 -->|"Consume"| C1["Consumer A"]
    Q2 -->|"Consume"| C2["Consumer B"]

RabbitMQ Exchange Type Characteristics #

RabbitMQ has 4 main Exchange types to control how messages are routed to queues:

Exchange TypeShort DescriptionRouting MechanismMain Use Cases
DirectDirect RoutingExact match between Routing Key and Binding KeySpecific task distribution
FanoutBroadcastSends to all bound queues regardless of KeyGlobal logging, broadcast notifications
TopicText PatternWildcard matching: * (1 word) and # (0 or more words)Event routing by structured category
HeadersHeader MatchingUses message headers instead of the Routing KeyFlexible routing based on metadata

Installation #

go get github.com/rabbitmq/amqp091-go

Connection and Channel #

import amqp "github.com/rabbitmq/amqp091-go"

func connect(url string) (*amqp.Connection, *amqp.Channel, error) {
    // Format: amqp://user:password@host:port/vhost
    conn, err := amqp.Dial(url)
    if err != nil {
        return nil, nil, fmt.Errorf("dial: %w", err)
    }

    ch, err := conn.Channel()
    if err != nil {
        conn.Close()
        return nil, nil, fmt.Errorf("channel: %w", err)
    }

    return conn, ch, nil
}

Resilient Connections (Reconnection) #

type RabbitMQ struct {
    url    string
    conn   *amqp.Connection
    ch     *amqp.Channel
    mu     sync.Mutex
    closed bool
}

func NewRabbitMQ(url string) (*RabbitMQ, error) {
    r := &RabbitMQ{url: url}
    if err := r.connect(); err != nil {
        return nil, err
    }
    go r.watchReconnect()
    return r, nil
}

func (r *RabbitMQ) connect() error {
    conn, err := amqp.Dial(r.url)
    if err != nil {
        return err
    }
    ch, err := conn.Channel()
    if err != nil {
        conn.Close()
        return err
    }
    r.conn = conn
    r.ch = ch
    return nil
}

func (r *RabbitMQ) watchReconnect() {
    for !r.closed {
        reason, ok := <-r.conn.NotifyClose(make(chan *amqp.Error))
        if !ok || r.closed {
            break
        }
        log.Printf("Connection lost: %v, attempting to reconnect...", reason)

        for {
            time.Sleep(5 * time.Second)
            if err := r.connect(); err != nil {
                log.Printf("Reconnect failed: %v, retrying...", err)
                continue
            }
            log.Println("Reconnect successful!")
            break
        }
    }
}

func (r *RabbitMQ) Close() {
    r.closed = true
    r.ch.Close()
    r.conn.Close()
}

Exchange and Queue Setup #

func setupDirectExchange(ch *amqp.Channel) error {
    // Declare the exchange
    err := ch.ExchangeDeclare(
        "orders",  // name
        "direct",  // type: direct, fanout, topic, headers
        true,      // durable: survives a restart
        false,     // auto-deleted
        false,     // internal
        false,     // no-wait
        nil,       // args
    )
    if err != nil {
        return fmt.Errorf("exchange declare: %w", err)
    }

    // Declare the queue
    q, err := ch.QueueDeclare(
        "order.processing",  // name (empty = random generated)
        true,                // durable
        false,               // delete when unused
        false,               // exclusive
        false,               // no-wait
        amqp.Table{
            "x-dead-letter-exchange":    "orders.dlx",   // dead letter exchange
            "x-dead-letter-routing-key": "order.failed", // routing key for the DLQ
            "x-message-ttl":             int32(3600000), // TTL 1 hour (ms)
        },
    )
    if err != nil {
        return fmt.Errorf("queue declare: %w", err)
    }

    // Bind the queue to the exchange with a routing key
    return ch.QueueBind(
        q.Name,             // queue
        "order.processing", // routing key
        "orders",           // exchange
        false,
        nil,
    )
}

// Fanout exchange — broadcast to all queues
func setupFanoutExchange(ch *amqp.Channel) error {
    err := ch.ExchangeDeclare("notifications", "fanout", true, false, false, false, nil)
    if err != nil {
        return err
    }

    // Each service has its own queue
    for _, service := range []string{"email", "sms", "push"} {
        q, err := ch.QueueDeclare(
            "notification."+service, true, false, false, false, nil)
        if err != nil {
            return err
        }
        // Fanout doesn't need a routing key
        if err := ch.QueueBind(q.Name, "", "notifications", false, nil); err != nil {
            return err
        }
    }
    return nil
}

// Topic exchange — routing with wildcards
func setupTopicExchange(ch *amqp.Channel) error {
    err := ch.ExchangeDeclare("events", "topic", true, false, false, false, nil)
    if err != nil {
        return err
    }

    queues := []struct {
        name       string
        routingKey string
    }{
        {"events.orders.all", "order.#"},          // all order events
        {"events.payments", "payment.*"},          // all payment events
        {"events.critical", "#.failed"},           // all failed events
        {"events.audit", "#"},                     // all events (audit log)
    }

    for _, q := range queues {
        queue, err := ch.QueueDeclare(q.name, true, false, false, false, nil)
        if err != nil {
            return err
        }
        if err := ch.QueueBind(queue.Name, q.routingKey, "events", false, nil); err != nil {
            return err
        }
    }
    return nil
}

Publish — Sending Messages #

func publish(ch *amqp.Channel, exchange, routingKey string, body interface{}) error {
    data, err := json.Marshal(body)
    if err != nil {
        return err
    }

    return ch.PublishWithContext(
        context.Background(),
        exchange,   // exchange
        routingKey, // routing key
        true,       // mandatory: return if no matching queue exists
        false,      // immediate
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent, // survives a broker restart
            MessageId:    uuid.New().String(),
            Timestamp:    time.Now(),
            Body:         data,
            Headers: amqp.Table{
                "source":  "order-service",
                "version": "1.0",
            },
        },
    )
}

// Publish with confirmation (Publisher Confirm)
func publishWithConfirm(ch *amqp.Channel, exchange, routingKey string, body interface{}) error {
    // Enable confirm mode
    if err := ch.Confirm(false); err != nil {
        return err
    }
    confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

    data, _ := json.Marshal(body)
    if err := ch.PublishWithContext(context.Background(), exchange, routingKey, false, false,
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,
            Body:         data,
        }); err != nil {
        return err
    }

    // Wait for the broker's confirmation
    confirm := <-confirms
    if !confirm.Ack {
        return fmt.Errorf("message not confirmed by the broker")
    }
    return nil
}

Consume — Reading Messages #

// Prefetch — limit the number of messages sent before acknowledgment
func startConsumer(ch *amqp.Channel, queueName string, handler func(amqp.Delivery) error) error {
    // Only send 5 messages at a time; wait for an ACK before sending more
    if err := ch.Qos(5, 0, false); err != nil {
        return fmt.Errorf("qos: %w", err)
    }

    msgs, err := ch.Consume(
        queueName, // queue
        "",        // consumer tag (empty = auto-generate)
        false,     // auto-ack (false = manual ack)
        false,     // exclusive
        false,     // no-local
        false,     // no-wait
        nil,
    )
    if err != nil {
        return fmt.Errorf("consume: %w", err)
    }

    go func() {
        for msg := range msgs {
            if err := handler(msg); err != nil {
                log.Printf("Error processing message: %v", err)
                // Nack with requeue=false → goes to the Dead Letter Queue
                msg.Nack(false, false)
                continue
            }
            // Ack — remove the message from the queue
            msg.Ack(false)
        }
    }()

    return nil
}

Dead Letter Queues (DLQ) #

Messages that fail processing (Nack without requeue) go to the DLQ for inspection or retry:

func setupDLQ(ch *amqp.Channel) error {
    // Exchange for the DLQ
    err := ch.ExchangeDeclare("orders.dlx", "direct", true, false, false, false, nil)
    if err != nil {
        return err
    }

    // Queue for failed messages
    _, err = ch.QueueDeclare("order.failed", true, false, false, false, nil)
    if err != nil {
        return err
    }

    return ch.QueueBind("order.failed", "order.failed", "orders.dlx", false, nil)
}

// Consumer for the DLQ — analysis and manual retry
func consumeDLQ(ch *amqp.Channel) error {
    return startConsumer(ch, "order.failed", func(msg amqp.Delivery) error {
        log.Printf("DLQ: MessageID=%s", msg.MessageId)
        log.Printf("  Headers: %v", msg.Headers)
        log.Printf("  Body: %s", string(msg.Body))

        // Analyze why it failed and decide: retry or discard
        retryCount, _ := msg.Headers["x-retry-count"].(int32)
        if retryCount < 3 {
            // Republish with the retry count incremented
            msg.Headers["x-retry-count"] = retryCount + 1
            return ch.PublishWithContext(context.Background(),
                "orders", "order.processing", false, false,
                amqp.Publishing{
                    ContentType:  msg.ContentType,
                    DeliveryMode: amqp.Persistent,
                    Headers:      msg.Headers,
                    Body:         msg.Body,
                })
        }

        // Already retried 3 times, save to permanent storage for manual review
        log.Printf("PERMANENT FAILURE: %s", msg.MessageId)
        return nil
    })
}

Complete Example Program — Notification System #

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

type Notification struct {
    ID        string    `json:"id"`
    Type      string    `json:"type"`    // email, sms, push
    UserID    string    `json:"user_id"`
    Subject   string    `json:"subject,omitempty"`
    Message   string    `json:"message"`
    Channel   string    `json:"channel"` // email, sms, push
    Timestamp time.Time `json:"timestamp"`
}

func setupRabbitMQ(ch *amqp.Channel) error {
    // Fanout exchange to broadcast notifications
    if err := ch.ExchangeDeclare(
        "notifications", "fanout", true, false, false, false, nil); err != nil {
        return err
    }

    // A queue for each notification channel
    channels := []string{"email", "sms", "push"}
    for _, name := range channels {
        q, err := ch.QueueDeclare(
            "notification."+name, true, false, false, false,
            amqp.Table{"x-message-ttl": int32(86400000)}, // TTL 24 hours
        )
        if err != nil {
            return err
        }
        if err := ch.QueueBind(q.Name, "", "notifications", false, nil); err != nil {
            return err
        }
    }
    return nil
}

func publishNotification(ch *amqp.Channel, notif Notification) error {
    data, _ := json.Marshal(notif)
    return ch.PublishWithContext(
        context.Background(),
        "notifications", "", false, false,
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,
            MessageId:    notif.ID,
            Timestamp:    notif.Timestamp,
            Body:         data,
        },
    )
}

func startNotificationWorker(ch *amqp.Channel, channelName string) error {
    ch.Qos(3, 0, false)

    msgs, err := ch.Consume(
        "notification."+channelName, "", false, false, false, false, nil)
    if err != nil {
        return err
    }

    go func() {
        log.Printf("[%s] Worker ready", channelName)
        for msg := range msgs {
            var notif Notification
            if err := json.Unmarshal(msg.Body, &notif); err != nil {
                log.Printf("[%s] ERROR unmarshal: %v", channelName, err)
                msg.Nack(false, false)
                continue
            }

            // Simulate sending based on the channel
            switch channelName {
            case "email":
                fmt.Printf("  📧 [EMAIL] to %s: %s\n", notif.UserID, notif.Subject)
                time.Sleep(50 * time.Millisecond)
            case "sms":
                fmt.Printf("  📱 [SMS] to %s: %s\n", notif.UserID, notif.Message[:min(30, len(notif.Message))]+"...")
                time.Sleep(30 * time.Millisecond)
            case "push":
                fmt.Printf("  🔔 [PUSH] to %s: %s\n", notif.UserID, notif.Message)
                time.Sleep(10 * time.Millisecond)
            }

            msg.Ack(false)
        }
    }()
    return nil
}

func min(a, b int) int {
    if a < b {
        return a
    }
    return b
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal("RabbitMQ connection:", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatal("Channel:", err)
    }
    defer ch.Close()

    if err := setupRabbitMQ(ch); err != nil {
        log.Fatal("Setup:", err)
    }
    fmt.Println("✓ RabbitMQ ready")

    // Start workers for each channel
    for _, name := range []string{"email", "sms", "push"} {
        if err := startNotificationWorker(ch, name); err != nil {
            log.Fatal("Worker:", err)
        }
    }

    // Publish some notifications
    fmt.Println("\n=== Sending Notifications ===")
    notifications := []Notification{
        {
            ID: "notif-001", UserID: "USR-42",
            Subject: "Order Confirmed",
            Message: "Order ORD-001 worth Rp1,500,000 has been confirmed.",
            Timestamp: time.Now(),
        },
        {
            ID: "notif-002", UserID: "USR-99",
            Subject: "Payment Successful",
            Message: "The payment for ORD-002 was received successfully.",
            Timestamp: time.Now(),
        },
        {
            ID: "notif-003", UserID: "USR-42",
            Subject: "Order Being Shipped",
            Message: "Order ORD-001 is on its way. Tracking no.: JNE123456.",
            Timestamp: time.Now(),
        },
    }

    for _, n := range notifications {
        if err := publishNotification(ch, n); err != nil {
            log.Printf("Failed to publish: %v", err)
        } else {
            fmt.Printf("  Sent: %s (to: %s)\n", n.ID, n.UserID)
        }
    }

    // Wait for the consumers to process
    fmt.Println("\n=== Processing Notifications ===")
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    select {
    case <-sigCh:
        fmt.Println("\nShutting down...")
    case <-time.After(3 * time.Second):
        fmt.Println("\nDone.")
    }
}

Summary #

  • amqp091-go is the official RabbitMQ library for Go — a fork of streadway/amqp, which is no longer maintained.
  • Exchange types: direct (exact routing key), fanout (broadcast), topic (wildcards * and #), headers.
  • DeliveryMode: amqp.Persistent is required for messages that must not be lost when the broker restarts.
  • Manual ACK (autoAck=false) with msg.Ack() after successful processing, msg.Nack(false, false) for the DLQ.
  • ch.Qos(n, 0, false) for prefetch — limits how many messages are sent before the worker ACKs.
  • Dead Letter Queues for failed messages — Nack without requeue → goes to the DLQ for analysis/retry.
  • Publisher Confirms guarantee messages reach the broker — enable with ch.Confirm(false).
  • Reconnection logic is important for production — monitor conn.NotifyClose() and reconnect automatically.
  • x-message-ttl for expiring unprocessed messages; x-dead-letter-exchange to route expired/failed messages.
  • Topic exchanges with # (zero or more words) and * (exactly one word) for very flexible routing.

← Previous: Kafka   Next: Amazon SQS →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact