rmq

RabbitMQ Go Library

A Client library for Go with built-in reconnection, retry logic, dead letter queue support, and worker pools.

Features

• Automatic reconnection with exponential backoff
• Message retry with configurable backoff strategies
• Dead Letter Queue (DLQ) support
• Worker pool for concurrent message processing
• Publisher confirms for reliable message delivery
• Channel pooling for efficient publishing
• Graceful shutdown handling
• Structured logging with log/slog

Installation

go get github.com/xorfall/rmq

Usage

Basic Setup

package main

import (
    "context"
    "log/slog"
    "os"
    "time"

    "github.com/xorfall/rmq"
)

func main() {
    // Setup logger
    logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
    }))

    ctx := context.Background()

    // Create RabbitMQ client
    client := rmq.NewClient(ctx, rmq.Config{
        URL:              "amqp://guest:guest@localhost:5672/",
        ReconnectWait:    time.Second,
        MaxReconnectWait: 30 * time.Second,
        DialTimeout:      5 * time.Second,
    }, logger)
    defer client.Close()

    // Wait for connection
    if err := client.WaitUntilReady(); err != nil {
        logger.Error("failed to connect", "error", err)
        os.Exit(1)
    }
}

Publishing Messages

// Create producer
producer := rmq.NewProducer(client, rmq.ProducerConfig{
    PoolSize:    5,
    UseConfirms: true,
}, logger)
defer producer.Close()

// Publish a message
err := producer.Publish(ctx, []byte("Hello World"), rmq.PublishOptions{
    Exchange:   "my-exchange",
    RoutingKey: "my-routing-key",
    Persistent: true,
    MessageID:  "msg-123",
})

Consuming Messages

// Create consumer
consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:         "my-queue",
    Exchange:      "my-exchange",
    RoutingKey:    "my-routing-key",
    PrefetchCount: 10,
    MaxRetries:    3,
    RetryBackoff: []time.Duration{
        1 * time.Second,
        5 * time.Second,
        15 * time.Second,
    },
    ProcessTimeout: 30 * time.Second,
    Workers:        3,
}, logger)

// Define message handler
handler := func(ctx context.Context, msg *rmq.Message) error {
    logger.Info("processing message", "id", msg.MessageID)
    // Process your message here
    return nil
}

// Define error handler (called when message fails after all retries)
errorHandler := func(ctx context.Context, msg *rmq.Message, err error) {
    logger.Error("message failed permanently", "id", msg.MessageID, "error", err)
}

// Start consumer
if err := consumer.Start(handler, errorHandler); err != nil {
    logger.Error("failed to start consumer", "error", err)
    os.Exit(1)
}
defer consumer.Stop()

Exchange Types

The library supports all RabbitMQ exchange types: direct, fanout, topic, and headers. You can configure the exchange type using the ExchangeType field in ConsumerConfig.

Direct Exchange (Default)

Routes messages to queues based on exact routing key match. Best for point-to-point communication.

consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "orders",
    Exchange:     "orders",
    ExchangeType: "direct",  // default
    RoutingKey:   "order.create",
}, logger)

Use cases: Task distribution, command routing, specific event handling

Fanout Exchange

Broadcasts messages to all bound queues, ignoring routing keys. Best for pub/sub patterns.

consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "notifications.email",
    Exchange:     "notifications",
    ExchangeType: "fanout",
    RoutingKey:   "",  // ignored for fanout
}, logger)

Use cases: Broadcasting notifications, cache invalidation, event replication

See examples/fanout for a complete example.

Topic Exchange

Routes messages based on wildcard pattern matching on routing keys. Best for flexible routing patterns.

Wildcard patterns:

• * matches exactly one word
• # matches zero or more words

// Consumer 1: All error logs from any service
consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "logs.errors",
    Exchange:     "logs",
    ExchangeType: "topic",
    RoutingKey:   "*.error",  // matches: auth.error, payment.error, etc.
}, logger)

// Consumer 2: All logs from auth service
consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "logs.auth",
    Exchange:     "logs",
    ExchangeType: "topic",
    RoutingKey:   "auth.*",  // matches: auth.info, auth.error, auth.warning
}, logger)

// Consumer 3: All logs
consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "logs.all",
    Exchange:     "logs",
    ExchangeType: "topic",
    RoutingKey:   "#",  // matches everything
}, logger)

Use cases: Log routing, hierarchical data, multi-criteria routing

See examples/topic for a complete example.

Headers Exchange

Routes messages based on header attributes instead of routing keys. Best for complex routing logic.

consumer := rmq.NewConsumer(client, rmq.ConsumerConfig{
    Queue:        "premium.orders",
    Exchange:     "orders",
    ExchangeType: "headers",
}, logger)

// When publishing, use headers for routing
producer.Publish(ctx, data, rmq.PublishOptions{
    Exchange: "orders",
    Headers: map[string]interface{}{
        "customer-tier": "premium",
        "region":        "us-east",
    },
})

Use cases: Complex routing conditions, attribute-based routing, multi-dimensional filtering

Note: Headers exchange requires manual queue binding with arguments via RabbitMQ management or AMQP commands, as Go's QueueBind doesn't support binding arguments directly.

Processing Dead Letter Queue (DLQ)

The library provides a DLQ processor to handle messages that have permanently failed. This allows you to:

• Inspect failed messages
• Retry messages after manual fixes
• Discard invalid messages
• Keep suspicious messages for investigation

// Create DLQ processor
dlqProcessor := rmq.NewDLQProcessor(client, rmq.DLQProcessorConfig{
    Queue:          "my-queue.dlq",      // DLQ name
    RetryQueue:     "my-queue",          // Original queue to retry to
    PrefetchCount:  5,
    ProcessTimeout: 30 * time.Second,
}, logger)

// Define DLQ handler - decides what to do with each failed message
dlqHandler := func(ctx context.Context, msg *rmq.Message) (rmq.DLQAction, error) {
    logger.Info("processing DLQ message", "id", msg.MessageID, "retry_count", msg.RetryCount)

    // Access error information from headers
    var lastError string
    if msg.Headers != nil {
        if err, ok := msg.Headers[rmq.HeaderErrorMsg].(string); ok {
            lastError = err
        }
    }

    // Decide action based on your business logic
    if strings.Contains(lastError, "temporary") {
        // Retry if it was a temporary error
        return rmq.DLQActionRetry, nil
    } else if strings.Contains(lastError, "invalid format") {
        // Discard messages with permanent errors
        return rmq.DLQActionDiscard, nil
    }

    // Keep in DLQ for manual review
    return rmq.DLQActionKeep, nil
}

// Start DLQ processor
if err := dlqProcessor.Start(dlqHandler); err != nil {
    logger.Error("failed to start DLQ processor", "error", err)
    os.Exit(1)
}
defer dlqProcessor.Stop()

DLQ Actions

The DLQ handler can return three different actions:

• DLQActionRetry: Requeues the message to the original queue with retry count reset

  • Use when: The issue has been fixed (e.g., external service is back online)
  • The message will be processed fresh, starting from retry count 0
  • Adds x-dlq-reprocessed: true header to track it was reprocessed from DLQ

• DLQActionDiscard: Permanently removes the message from DLQ

  • Use when: The message is invalid and cannot be processed
  • Message is acknowledged and deleted
  • Make sure to log important information before discarding

• DLQActionKeep: Leaves the message in the DLQ

  • Use when: Manual investigation is needed
  • Message is nacked without requeue
  • Message stays in DLQ for later processing

Configuration

Client Config

• URL: RabbitMQ connection URL (e.g., "amqp://user:pass@host:port/vhost")
• ReconnectWait: Initial wait time before reconnection (default: 1s)
• MaxReconnectWait: Maximum wait time between reconnection attempts (default: 60s)
• DialTimeout: Timeout for establishing connection (default: 5s)

Producer Config

• PoolSize: Number of channels in the pool (default: 5)
• UseConfirms: Enable publisher confirms for reliability (default: false)

Consumer Config

• Queue: Queue name to consume from
• Exchange: Exchange to bind the queue to
• RoutingKey: Routing key for queue binding
• ExchangeType: Exchange type - "direct", "fanout", "topic", or "headers" (default: "direct")
• PrefetchCount: Number of messages to prefetch (default: 10)
• MaxRetries: Maximum retry attempts before sending to DLQ (default: 3)
• RetryBackoff: Backoff duration for each retry attempt
• DeadLetterExchange: DLX name (default: "{Exchange}.dlx")
• DeadLetterRoutingKey: DLQ routing key (default: "{Queue}.dead")
• ProcessTimeout: Timeout for processing a single message (default: 30s)
• Workers: Number of concurrent workers (default: 1)

Retry Backoff Patterns

The library supports flexible retry backoff patterns. You can either specify delays manually or use built-in helper functions:

Manual Array (Full Control):

RetryBackoff: []time.Duration{1*time.Second, 5*time.Second, 15*time.Second}

Exponential Backoff:

RetryBackoff: rmq.ExponentialBackoff(
    1*time.Second,   // initial delay
    2.0,             // multiplier
    30*time.Second,  // max delay
    5,               // retry count
)
// Results: [1s, 2s, 4s, 8s, 16s]

Exponential with Jitter (Recommended for Distributed Systems):

RetryBackoff: rmq.ExponentialBackoffWithJitter(
    1*time.Second,   // initial delay
    2.0,             // multiplier
    30*time.Second,  // max delay
    5,               // retry count
    0.3,             // 30% jitter
)
// Results: [~0.7-1.3s, ~1.4-2.6s, ~2.8-5.2s, ~5.6-10.4s, ~11.2-20.8s]

Why Jitter? Jitter adds randomness to retry delays, preventing the "thundering herd" problem where many failed messages retry at exactly the same time, potentially overwhelming the system again. Recommended for production systems with multiple consumers.

Linear Backoff:

RetryBackoff: rmq.LinearBackoff(1*time.Second, 2*time.Second, 5)
// Results: [1s, 3s, 5s, 7s, 9s]

Constant Backoff:

RetryBackoff: rmq.ConstantBackoff(5*time.Second, 3)
// Results: [5s, 5s, 5s]

DLQ Processor Config

• Queue: DLQ name to process (e.g., "my-queue.dlq")
• RetryQueue: Original queue name to retry messages to (e.g., "my-queue")
• PrefetchCount: Number of messages to prefetch from DLQ (default: 5)
• ProcessTimeout: Timeout for processing a single DLQ message (default: 30s)

Examples

The library includes four comprehensive examples:

1. Basic Example

See examples/basic for a complete example showing:

• Basic producer and consumer setup
• Message retry with backoff
• Error handling
• Graceful shutdown

# Start RabbitMQ (using Docker)
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# Run the basic example
cd examples/basic
go run main.go

2. DLQ Example

See examples/dlq for a complete example demonstrating:

• Messages failing and going to DLQ
• DLQ processor handling dead-lettered messages
• Different DLQ actions (Retry, Discard, Keep)
• Error metadata tracking
• Conditional message reprocessing

# Start RabbitMQ (using Docker)
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# Run the DLQ example
cd examples/dlq
go run main.go

What the DLQ example demonstrates:

• PAY-001: Processes successfully
• PAY-002: Fails first time, succeeds on retry (temporary failure)
• PAY-003: Goes to DLQ after retries, then gets retried from DLQ
• PAY-004: Goes to DLQ and stays there (fraud case)
• PAY-005: Processes successfully

3. Topic Exchange Example

See examples/topic for a complete example demonstrating:

• Topic exchange with wildcard routing patterns
• Multiple consumers with different routing key patterns
• Pattern matching with * (one word) and # (zero or more words)
• Log routing by service and level

# Run the topic example
cd examples/topic
go run main.go

4. Fanout Exchange Example

See examples/fanout for a complete example demonstrating:

• Fanout exchange broadcasting to all queues
• Multiple consumers receiving the same messages
• Notification delivery via different channels (email, SMS, push)

# Run the fanout example
cd examples/fanout
go run main.go

How It Works

Retry Mechanism

When a message fails to process:

1. The library checks if max retries has been reached
2. If not, it republishes the message with an incremented retry count
3. A backoff delay is applied before the message is reprocessed
4. Retry metadata is stored in message headers

Dead Letter Queue (DLQ)

The Dead Letter Queue (DLQ) is a safety mechanism for handling messages that cannot be processed successfully after multiple retry attempts.

Message Flow

Here's how a message flows through the system when it encounters errors:

1. Message published to queue
   ↓
2. Consumer attempts to process
   ↓
3. Processing fails → Error returned
   ↓
4. [RETRY LOGIC]
   - Check if retry count < MaxRetries
   - If YES: Apply backoff delay → Republish to same queue (with incremented retry count)
   - If NO: Proceed to step 5
   ↓
5. Max retries exhausted
   ↓
6. ErrorHandler callback invoked (for logging, alerting, etc.)
   ↓
7. Message published to Dead Letter Exchange (DLX)
   ↓
8. Message routed to Dead Letter Queue (DLQ)
   ↓
9. DLQ Processor handles the message (optional)
   - Inspect message and error details
   - Decide action: Retry, Discard, or Keep

Automatic DLQ Setup

When you create a consumer, the library automatically:

1. Creates Dead Letter Exchange (DLX)

   • Name: {Exchange}.dlx (e.g., "orders.dlx")
   • Type: Direct exchange
   • Durable: Yes

2. Creates Dead Letter Queue

   • Name: {Queue}.dlq (e.g., "orders.process.dlq")
   • Durable: Yes
   • Bound to DLX with routing key: {Queue}.dead

3. Configures Main Queue with DLX

   • Sets x-dead-letter-exchange argument
   • Sets x-dead-letter-routing-key argument
   • Messages that are rejected/nacked go to DLX

Message Headers in DLQ

Messages in the DLQ contain helpful metadata in their headers:

• x-retry-count (int32): Number of times the message was retried
• x-first-error (int64): Unix timestamp of first failure
• x-last-error (int64): Unix timestamp of last failure
• x-error-message (string): The error message from the last failure
• x-dlq-reprocessed (bool): Set to true when retried from DLQ
• x-dlq-reprocessed-at (int64): Unix timestamp when reprocessed from DLQ

Access these in your handler:

func dlqHandler(ctx context.Context, msg *rmq.Message) (rmq.DLQAction, error) {
    // Access retry count
    retryCount := msg.RetryCount

    // Access error information
    var errorMsg string
    if msg.Headers != nil {
        if err, ok := msg.Headers[rmq.HeaderErrorMsg].(string); ok {
            errorMsg = err
        }
    }

    // Access timestamps
    firstError := msg.FirstError // time.Time

    // Make decision based on this data
    // ...
}

DLQ Processing Strategies

1. Immediate Retry - Retry as soon as issue is resolved

// After deploying a bug fix, retry all messages
return rmq.DLQActionRetry, nil

2. Conditional Retry - Retry based on error type

if strings.Contains(errorMsg, "timeout") || strings.Contains(errorMsg, "unavailable") {
    // External service might be back up
    return rmq.DLQActionRetry, nil
}
return rmq.DLQActionKeep, nil

3. Time-based Retry - Retry after certain time period

if time.Since(msg.FirstError) > 24*time.Hour {
    // It's been a day, the issue might be resolved
    return rmq.DLQActionRetry, nil
}
return rmq.DLQActionKeep, nil

4. Alert and Keep - Notify team and keep for investigation

if msg.RetryCount > 5 {
    // Send alert to ops team
    sendAlert(msg)
    return rmq.DLQActionKeep, nil
}
return rmq.DLQActionDiscard, nil

5. Selective Discard - Remove messages that can't be fixed

// Parse message to check if it's fixable
var payment Payment
json.Unmarshal(msg.Body, &payment)

if payment.IsValid() {
    return rmq.DLQActionRetry, nil
}
// Invalid data, cannot be fixed
return rmq.DLQActionDiscard, nil

Best Practices

1. Always Log Before Discarding

   if action == rmq.DLQActionDiscard {
       logger.Error("discarding message",
           "message_id", msg.MessageID,
           "body", string(msg.Body),
           "error", errorMsg)
   }
   return rmq.DLQActionDiscard, nil
   

2. Set Alerts for DLQ Growth

   • Monitor DLQ depth
   • Alert when messages accumulate
   • Investigate root causes

3. Implement DLQ Processor Carefully

   • Start with DLQActionKeep as default
   • Only auto-retry when you're confident
   • Avoid infinite retry loops

4. Track Reprocessed Messages

   • Check for x-dlq-reprocessed header
   • Limit number of DLQ reprocessing attempts
   • Prevent messages from bouncing between queue and DLQ

5. Use ErrorHandler for Immediate Actions

   errorHandler := func(ctx context.Context, msg *rmq.Message, err error) {
       // Send immediate alert
       alerting.Send("Message failed permanently", msg.MessageID)
   
       // Update database
       db.UpdateStatus(msg.MessageID, "failed")
   
       // Create support ticket
       ticketing.Create(msg, err)
   }
   

Automatic Reconnection

The client automatically handles connection failures:

1. Detects connection loss
2. Attempts reconnection with exponential backoff
3. Consumers automatically resume consuming after reconnection
4. In-flight messages are handled gracefully

Project Structure

.
├── client.go           # RabbitMQ client with reconnection
├── consumer.go         # Consumer with retry and DLQ
├── producer.go         # Producer with channel pooling
├── types.go            # Shared types and configs
├── dlq_processor.go    # DLQ processing utilities
├── examples/
│   ├── basic/
│   │   └── main.go     # Basic producer/consumer example
│   ├── dlq/
│   │   └── main.go     # DLQ processing example
│   ├── topic/
│   │   └── main.go     # Topic exchange example
│   └── fanout/
│       └── main.go     # Fanout exchange example
└── README.md

License

MIT