Laravel RabbitMQ

A production-ready RabbitMQ queue driver for Laravel with attribute-based topology, automatic retries, and Kubernetes-native deployment.

Build resilient, scalable queue systems using RabbitMQ's powerful routing with Laravel's familiar job syntax. No Horizon required.

✨ Features

🎯 Attribute-Based Topology - Define exchanges and queues using PHP 8 attributes on your job classes
🔄 Automatic Retries & DLQ - Built-in dead letter queues with configurable retry strategies
📊 Priority Queues - Support for message priorities (0-255) on classic queues
⏰ Delayed Messages - Schedule jobs with native RabbitMQ delayed message exchange
🚀 Laravel-Native - Works with standard dispatch() - no learning curve
☸️ Kubernetes-Ready - Custom consumer commands designed for containerized deployments
💪 Production-Proven - Built on php-amqplib with heartbeat support and publisher confirms

🤔 Why Use RabbitMQ?

This package is ideal for applications that need:

Advanced Routing - Route messages based on patterns, headers, or broadcast to multiple queues
Guaranteed Delivery - RabbitMQ's persistence and publisher confirms ensure messages aren't lost
Complex Workflows - Multi-tenant systems, event-driven architectures, microservices communication
Infrastructure-Level Control - Manage queue topology, clustering, and federation through RabbitMQ itself
Kubernetes-Native Workers - Deploy queue consumers as standard Kubernetes Deployments with HPA
Protocol Flexibility - AMQP protocol support for cross-platform messaging (Node.js, Python, Go, etc.)

📋 Requirements

PHP 8.2+
Laravel 11.0+ or 12.0+
RabbitMQ 3.12+
php-amqplib/php-amqplib ^3.6

Optional:

rabbitmq_delayed_message_exchange plugin for delayed messages
rabbitmq_prometheus plugin for Prometheus metrics

📦 Installation

composer require lettermint/laravel-rabbitmq

Publish the configuration file:

php artisan vendor:publish --tag=rabbitmq-config

Update your config/queue.php:

'connections' => [
    'rabbitmq' => [
        'driver' => 'rabbitmq',
        'queue' => env('RABBITMQ_QUEUE', 'default'),
        'exchange' => env('RABBITMQ_EXCHANGE', ''),
    ],
],

// Set as default if desired
'default' => env('QUEUE_CONNECTION', 'rabbitmq'),

Add to your .env:

QUEUE_CONNECTION=rabbitmq
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

🚀 Quick Start

Here's a complete example - from defining a job to processing it:

1. Define Your Job

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Lettermint\RabbitMQ\Attributes\ConsumesQueue;

#[ConsumesQueue(
    queue: 'emails',
    bindings: ['notifications' => 'email.*'],  // Listens to notifications exchange
    quorum: true,                               // High availability queue
    retryAttempts: 3,                           // Retry up to 3 times
)]
class SendEmailJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, SerializesModels;

    public function __construct(
        public string $email,
        public string $subject,
        public string $message,
    ) {}

    public function handle(): void
    {
        // Send your email
        Mail::to($this->email)->send(new NotificationMail($this->subject, $this->message));
    }
}

2. Declare Topology

# Create the exchange, queue, and bindings in RabbitMQ
php artisan rabbitmq:declare

3. Dispatch & Consume

// Dispatch the job (from anywhere in your app)
SendEmailJob::dispatch('user@example.com', 'Welcome!', 'Thanks for signing up');

# Start consuming (in production, run this in a container/supervisor)
php artisan rabbitmq:consume emails

That's it! Your job will be routed through RabbitMQ and processed with automatic retries and dead letter handling.

💡 Tip: In production, run consumers as Kubernetes Deployments or supervisor processes.

📚 Core Concepts

Understanding these RabbitMQ concepts will help you use this package effectively:

Exchange → Binding → Queue Flow

┌─────────────┐    routing key: email.welcome     ┌──────────────┐
│   Producer  │──────────────────────────────────►│   Exchange   │
└─────────────┘                                   │ "notifications"│
                                                  └───────┬────────┘
                                                          │ binding: email.*
                                                          ▼
                                                  ┌──────────────┐
                                                  │    Queue     │
                                                  │   "emails"   │
                                                  └───────┬──────┘
                                                          │
                                                          ▼
                                                  ┌──────────────┐
                                                  │   Consumer   │
                                                  │ (Your Job)   │
                                                  └──────────────┘

Exchange: Routes messages based on routing keys (like a post office)
Queue: Stores messages until consumed (like a mailbox)
Binding: Routing rule connecting exchange to queue (e.g., email.* matches email.welcome)
Routing Key: Label on each message determining which queue(s) receive it

Attribute-Based Configuration

Instead of manually configuring exchanges and queues in RabbitMQ, define them with attributes:

// This attribute tells the package:
// 1. Create a queue named "emails"
// 2. Create a quorum queue (HA, durable)
// 3. Bind it to the "notifications" exchange with pattern "email.*"
// 4. Set up DLQ with 3 retry attempts
#[ConsumesQueue(
    queue: 'emails',
    bindings: ['notifications' => 'email.*'],
    quorum: true,
    retryAttempts: 3,
)]

When you run php artisan rabbitmq:declare, the package scans your job classes and creates everything automatically.

📖 Usage Examples

Basic Usage

Simple job with a queue:

#[ConsumesQueue(
    queue: 'default',
    bindings: ['tasks' => '#'],  // Catch all messages from 'tasks' exchange
)]
class ProcessTaskJob implements ShouldQueue
{
    public function handle(): void
    {
        // Process the task
    }
}

Creating an exchange (optional - useful for organization):

<?php

namespace App\RabbitMQ\Exchanges;

use Lettermint\RabbitMQ\Attributes\Exchange;
use Lettermint\RabbitMQ\Enums\ExchangeType;

#[Exchange(name: 'tasks', type: ExchangeType::Topic)]
class TasksExchange {}

Intermediate Usage

Priority queues for time-sensitive jobs:

use Lettermint\RabbitMQ\Contracts\HasPriority;

#[ConsumesQueue(
    queue: 'urgent-tasks',
    bindings: ['tasks' => 'urgent.*'],
    quorum: false,        // Priority requires classic queue
    maxPriority: 10,      // 0 = lowest, 10 = highest
)]
class UrgentTaskJob implements ShouldQueue, HasPriority
{
    public function __construct(
        public string $taskId,
        public int $priority = 5,
    ) {}

    public function getPriority(): int
    {
        return $this->priority;
    }
}

// Dispatch with high priority
UrgentTaskJob::dispatch($taskId, priority: 10);

Delayed/scheduled messages:

// Requires rabbitmq_delayed_message_exchange plugin
// Enable in config/rabbitmq.php: 'delayed.enabled' => true

// Delay by seconds
SendEmailJob::dispatch($email)->delay(300);  // 5 minutes

// Delay with Carbon
SendEmailJob::dispatch($email)->delay(now()->addHours(2));

// Schedule for specific time
SendEmailJob::dispatch($email)->delay(now()->tomorrow()->setHour(9));

Advanced Usage

Multiple bindings (listen to multiple routing patterns):

#[ConsumesQueue(
    queue: 'notifications',
    bindings: [
        'events' => ['user.created', 'user.updated'],
        'alerts' => 'critical.*',
    ],
)]

Exchange-to-exchange binding (hierarchical routing):

// Parent exchange
#[Exchange(name: 'events', type: ExchangeType::Topic)]
class EventsExchange {}

// Child exchange bound to parent
#[Exchange(
    name: 'user-events',
    type: ExchangeType::Topic,
    bindTo: 'events',
    bindRoutingKey: 'user.#',
)]
class UserEventsExchange {}

Custom retry strategy:

use Lettermint\RabbitMQ\Enums\RetryStrategy;

#[ConsumesQueue(
    queue: 'api-calls',
    bindings: ['tasks' => 'api.*'],
    retryAttempts: 5,
    retryStrategy: RetryStrategy::Exponential,
    retryDelays: [30, 60, 300, 900, 3600],  // 30s, 1m, 5m, 15m, 1h
)]

Repeatable attribute (one job, multiple queues):

// This job can be consumed from either queue
#[ConsumesQueue(queue: 'primary', bindings: ['tasks' => 'important.*'])]
#[ConsumesQueue(queue: 'secondary', bindings: ['tasks' => 'background.*'])]
class FlexibleJob implements ShouldQueue
{
    // ...
}

⚙️ Configuration Reference

Environment Variables

# Connection
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

# Behavior
RABBITMQ_HEARTBEAT=60
RABBITMQ_PREFETCH_COUNT=10
RABBITMQ_DELAYED_ENABLED=true

# Publisher
RABBITMQ_PUBLISHER_CONFIRM=true

Attribute: `#[Exchange]`

#[Exchange(
    name: 'events',                    // Required: Exchange name
    type: ExchangeType::Topic,         // topic, direct, fanout, headers, x-delayed-message
    durable: true,                     // Survive broker restart
    autoDelete: false,                 // Delete when no bindings
    internal: false,                   // Only accessible via e2e bindings
    bindTo: 'parent-exchange',         // Parent exchange for e2e binding
    bindRoutingKey: 'events.#',        // Routing pattern for parent
    arguments: [],                     // Custom exchange arguments
)]

Exchange Types:

Topic: Pattern-based routing (e.g., user.*.created)
Direct: Exact routing key match
Fanout: Broadcast to all bound queues
Headers: Route by message headers
DelayedMessage: Delayed delivery (requires plugin)

Attribute: `#[ConsumesQueue]`

#[ConsumesQueue(
    // Required
    queue: 'my-queue',                           // Queue name

    // Bindings
    bindings: [                                  // Exchange => routing key(s)
        'exchange-name' => 'routing.key',
        'other-exchange' => ['key1', 'key2'],
    ],

    // Queue type (choose one)
    quorum: true,                                // Quorum queue (HA, recommended)
    maxPriority: 10,                             // Classic with priority (quorum: false)

    // Limits
    messageTtl: 86400000,                        // Message TTL in ms (24h)
    maxLength: 1000000,                          // Max queue length
    overflow: OverflowBehavior::RejectPublishDlx, // Overflow behavior

    // Dead letter & retry
    dlqExchange: null,                           // Custom DLQ exchange (auto-derived if null)
    retryAttempts: 3,                            // Max retries before permanent DLQ
    retryStrategy: RetryStrategy::Exponential,   // exponential, fixed, linear
    retryDelays: [60, 300, 900],                 // Delays in seconds

    // Consumer settings
    prefetch: 10,                                // Messages to prefetch (QoS)
    timeout: 30,                                 // Job timeout in seconds
)]

Important Notes:

Quorum queues provide high availability but don't support priorities
Use classic queues (quorum: false) if you need maxPriority
DLQ exchange is auto-created based on your first binding exchange

Config File Options

See config/rabbitmq.php for full options. Key settings:

// Discovery paths (where to scan for attributes)
'discovery' => [
    'paths' => [
        app_path('Jobs'),
        app_path('RabbitMQ'),
    ],
],

// Default queue settings (for jobs without attributes)
'queue' => [
    'exchange' => '',  // Fallback exchange
],

// Delayed messages
'delayed' => [
    'enabled' => true,
    'max_delay' => 86400000,  // 24 hours max
],

🎮 Artisan Commands

Topology Management

# Declare all exchanges, queues, and bindings
php artisan rabbitmq:declare

# Preview what will be created (dry run)
php artisan rabbitmq:declare --dry-run

# View topology as tree
php artisan rabbitmq:topology

# Export topology as JSON
php artisan rabbitmq:topology --format=json

Queue Operations

# List all queues with stats
php artisan rabbitmq:queues

# Include DLQ queues in list
php artisan rabbitmq:queues --include-dlq

# Watch mode (updates every 2s)
php artisan rabbitmq:queues --watch

# Purge a queue (delete all messages)
php artisan rabbitmq:purge my-queue

Consumer

# Start consuming from a queue
php artisan rabbitmq:consume my-queue

# With custom settings
php artisan rabbitmq:consume my-queue \
    --prefetch=25 \
    --timeout=120 \
    --max-jobs=500 \
    --max-memory=256

# Stop when empty (useful for testing)
php artisan rabbitmq:consume my-queue --stop-when-empty

Consumer Options:

--prefetch: Messages to prefetch (default: 10)
--timeout: Job timeout in seconds (default: 60)
--max-jobs: Exit after N jobs (0 = unlimited)
--max-time: Exit after N seconds (0 = unlimited)
--max-memory: Exit if memory exceeds N MB (default: 128)
--stop-when-empty: Exit when queue is empty

Dead Letter Queue Operations

# Replay DLQ messages back to original queue
php artisan rabbitmq:replay-dlq my-queue

# Preview replay without moving messages
php artisan rabbitmq:replay-dlq my-queue --dry-run

# Limit number of messages to replay
php artisan rabbitmq:replay-dlq my-queue --limit=100

# Inspect DLQ messages without removing them
php artisan rabbitmq:dlq-inspect my-queue

# Inspect specific message
php artisan rabbitmq:dlq-inspect my-queue --id=message-uuid

# Limit number of messages shown
php artisan rabbitmq:dlq-inspect my-queue --limit=20

# JSON output
php artisan rabbitmq:dlq-inspect my-queue --format=json

# Purge DLQ messages (permanently delete)
php artisan rabbitmq:dlq-purge my-queue

# Purge specific message
php artisan rabbitmq:dlq-purge my-queue --id=message-uuid

# Purge old messages only
php artisan rabbitmq:dlq-purge my-queue --older-than=7d

# Preview without deleting
php artisan rabbitmq:dlq-purge my-queue --dry-run

# Skip confirmation
php artisan rabbitmq:dlq-purge my-queue --force

Monitoring

# Health check
php artisan rabbitmq:health

# JSON output (for monitoring tools)
php artisan rabbitmq:health --json

🔄 Dead Letter Queues

DLQs are automatically created for every queue to handle failed messages.

How It Works

┌──────────────┐
│ Original Queue│  Job fails or times out
│  "emails"    │─────────────┐
└──────────────┘             │
                             ▼
                     ┌──────────────┐
                     │  DLQ Exchange│
                     │ "notifications.dlq"
                     └───────┬──────┘
                             │
                             ▼
                     ┌──────────────┐
                     │   DLQ Queue  │  Retry after delay
                     │ "dlq:emails" │─────────────┐
                     └──────────────┘             │
                                                  │
         ┌────────────────────────────────────────┘
         │
         ▼
┌──────────────┐
│ Original Queue│  If retries exhausted → stays in DLQ
│  "emails"    │
└──────────────┘

Retry Strategies

Exponential (recommended for API calls):

retryStrategy: RetryStrategy::Exponential,
retryDelays: [60, 300, 900],  // 1m, 5m, 15m, then 15m for remaining

Fixed (same delay every time):

retryStrategy: RetryStrategy::Fixed,
retryDelays: [300],  // Always 5 minutes

Linear (increasing delay):

retryStrategy: RetryStrategy::Linear,
retryDelays: [60],  // 1m, 2m, 3m, 4m...

Messages Go to DLQ When:

Job throws unhandled exception (after retries)
Job exceeds timeout
Consumer rejects without requeue
Queue message TTL expires
Queue max-length exceeded (with overflow: RejectPublishDlx)

☸️ Kubernetes Deployment

Deploy workers as Kubernetes Deployments for automatic scaling and restarts.

Basic Worker Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: queue-worker-emails
spec:
  replicas: 3
  selector:
    matchLabels:
      app: queue-worker
      queue: emails
  template:
    metadata:
      labels:
        app: queue-worker
        queue: emails
    spec:
      containers:
        - name: worker
          image: your-app:latest
          command: ["php", "artisan", "rabbitmq:consume", "emails"]
          args:
            - "--prefetch=25"
            - "--max-jobs=500"
            - "--max-memory=256"
          env:
            - name: RABBITMQ_HOST
              value: "rabbitmq.default.svc.cluster.local"
            - name: RABBITMQ_USER
              valueFrom:
                secretKeyRef:
                  name: rabbitmq-credentials
                  key: username
            - name: RABBITMQ_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: rabbitmq-credentials
                  key: password
          resources:
            requests:
              memory: "128Mi"
              cpu: "100m"
            limits:
              memory: "512Mi"
              cpu: "500m"

Horizontal Pod Autoscaler (HPA)

Scale based on queue depth using KEDA or RabbitMQ metrics:

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: queue-worker-emails-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: queue-worker-emails
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: External
      external:
        metric:
          name: rabbitmq_queue_messages_ready
          selector:
            matchLabels:
              queue: emails
        target:
          type: AverageValue
          averageValue: "100"  # Target: 100 messages per pod

Best Practices

Set --max-jobs to restart workers periodically (prevents memory leaks)
Set --max-memory slightly below container limits
Use livenessProbe and readinessProbe for health checks
Run rabbitmq:declare in init container or CI/CD pipeline
Use PodDisruptionBudget to maintain availability during updates

🔧 Advanced Topics

Fallback Routing for Third-Party Jobs

Jobs without #[ConsumesQueue] (e.g., from packages) use fallback routing:

Routing: config('rabbitmq.queue.exchange') with key 'fallback.{queue_name}'

Create a catch-all queue for these:

#[ConsumesQueue(
    queue: 'fallback',
    bindings: ['your-exchange' => 'fallback.#'],
)]
class FallbackJob implements ShouldQueue {}

Quorum vs Classic Queues

Use Quorum Queues (default) when:

You need high availability (HA)
Data durability is critical
Running in clustered RabbitMQ

Use Classic Queues when:

You need message priorities
You need very low latency (single-node)
Legacy compatibility required

Cannot combine: quorum: true and maxPriority are mutually exclusive.

Publisher Confirms

Publisher confirms ensure messages reach RabbitMQ successfully. Enabled by default:

'publisher' => [
    'confirm' => true,  // Wait for RabbitMQ acknowledgment
],

If confirm fails, Laravel throws an exception and the job can be retried by your queue worker.

Heartbeats & Long-Running Jobs

The package sends heartbeats automatically during job execution to prevent connection timeouts.

For jobs longer than 2× heartbeat interval:

Heartbeats work automatically with ext-pcntl
If job has $timeout property, heartbeats are disabled during execution (both use SIGALRM)
For long jobs needing heartbeat: set public $timeout = 0; on the job class

🐛 Troubleshooting

Connection Issues

Problem: AMQPConnectionException: Connection refused

Solutions:

Verify RabbitMQ is running: docker ps or systemctl status rabbitmq-server
Check connection details in .env match your RabbitMQ instance
Ensure firewall allows port 5672
Test connection: telnet rabbitmq-host 5672

Messages Not Routing

Problem: Messages published but not appearing in queue

Solutions:

Run php artisan rabbitmq:topology to verify bindings
Check routing key matches binding pattern:
- email.* matches email.welcome but not email.welcome.urgent
- email.# matches email.welcome.urgent
Verify exchange and queue were declared: php artisan rabbitmq:declare
Check RabbitMQ management UI (port 15672) for unrouted messages

Consumer Stops Unexpectedly

Problem: rabbitmq:consume exits without error

Solutions:

Check memory limit: --max-memory=256 (increase if needed)
Check job limit: --max-jobs=500 (consumer exits after N jobs by design)
Check time limit: --max-time=3600 (consumer exits after N seconds)
Review logs for connection errors or exceptions
Verify heartbeat settings if jobs run longer than 2× heartbeat interval

Jobs Fail Silently

Problem: Jobs marked as processed but work not completed

Solutions:

Check your job's handle() method for unhandled exceptions
Enable failed job logging: check failed_jobs table
Review RabbitMQ DLQ: php artisan rabbitmq:dlq-inspect your-queue
Add logging to job: Log::info('Job started', ['id' => $this->id]);

Priority Not Working

Problem: High priority messages not processed first

Solutions:

Verify quorum: false (quorum queues don't support priority)
Verify maxPriority is set on queue attribute
Ensure job implements HasPriority interface
Check messages have priority set before prefetched messages processed

Delayed Messages Not Working

Problem: ->delay() doesn't delay message

Solutions:

Install plugin: rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Verify enabled in config: 'delayed.enabled' => true
Run php artisan rabbitmq:declare to create delayed exchange
Check delay is within max: default 24 hours (delayed.max_delay)

High Memory Usage

Problem: Worker memory grows over time

Solutions:

Set --max-memory=256 to restart worker before OOM
Set --max-jobs=500 to periodically restart workers
Check for memory leaks in job code
Ensure job releases large objects: unset($largeVariable);
Use --max-time=3600 for time-based restarts

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md for details.

📄 License

MIT License. See LICENSE for details.

Name		Name	Last commit message	Last commit date
Latest commit History 55 Commits
.github		.github
config		config
src		src
tests		tests
.editorconfig		.editorconfig
.gitattributes		.gitattributes
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE.md		LICENSE.md
README.md		README.md
composer.json		composer.json
composer.lock		composer.lock
phpstan-baseline.neon		phpstan-baseline.neon
phpstan.neon.dist		phpstan.neon.dist
phpunit.xml.dist		phpunit.xml.dist

Folders and files

Latest commit

History

Repository files navigation

Laravel RabbitMQ

✨ Features

🤔 Why Use RabbitMQ?

📋 Requirements

📦 Installation

🚀 Quick Start

1. Define Your Job

2. Declare Topology

3. Dispatch & Consume

📚 Core Concepts

Exchange → Binding → Queue Flow

Attribute-Based Configuration

📖 Usage Examples

Basic Usage

Intermediate Usage

Advanced Usage

⚙️ Configuration Reference

Environment Variables

Attribute: #[Exchange]

Attribute: #[ConsumesQueue]

Config File Options

🎮 Artisan Commands

Topology Management

Queue Operations

Consumer

Dead Letter Queue Operations

Monitoring

🔄 Dead Letter Queues

How It Works

Retry Strategies

Messages Go to DLQ When:

☸️ Kubernetes Deployment

Basic Worker Deployment

Horizontal Pod Autoscaler (HPA)

Best Practices

🔧 Advanced Topics

Fallback Routing for Third-Party Jobs

Quorum vs Classic Queues

Publisher Confirms

Heartbeats & Long-Running Jobs

🐛 Troubleshooting

Connection Issues

Messages Not Routing

Consumer Stops Unexpectedly

Jobs Fail Silently

Priority Not Working

Delayed Messages Not Working

High Memory Usage

🤝 Contributing

📄 License

About

Resources

License

Contributing

Uh oh!

Stars

Watchers

Forks

Releases 4

Uh oh!

Contributors

Uh oh!

Languages

Attribute: `#[Exchange]`

Attribute: `#[ConsumesQueue]`