Laravel RabbitMQ

A robust RabbitMQ queue driver for Laravel featuring native Laravel Queue API integration, advanced connection pooling, channel management, and high-performance async processing capabilities.

✨ Key Features

Core Features

🚀 High-Performance Connection Pooling - Reuse connections efficiently with configurable limits
🔄 Advanced Channel Management - Optimal AMQP channel pooling and lifecycle management
🛡️ Resilient Retry Strategy - Exponential backoff with configurable retry attempts
📊 Real-time Pool Monitoring - Built-in statistics and health monitoring
🔧 Production Ready - Comprehensive error handling and graceful shutdowns
⚡ Laravel Native - Seamless integration with Laravel Queue API
🎯 Zero Configuration - Works out of the box with sensible defaults

Advanced Features

🔀 Advanced Routing - Topic, fanout, and headers exchanges for complex routing patterns
🎯 Multi-Queue Support - Configure and manage multiple queues with different settings
📈 Exponential Backoff - Intelligent retry mechanism with jitter to prevent thundering herd
💀 Dead Letter Exchange - Automatic handling of failed messages with DLX
🔁 RPC Support - Synchronous request-response pattern for remote procedure calls
✅ Publisher Confirms - Reliable message delivery with broker acknowledgments
🔒 Transaction Management - Atomic operations with commit/rollback support
⏰ Delayed Messages - Schedule messages for future delivery (TTL or plugin-based)
🐌 Lazy Queues - Optimize memory usage for high-volume message queues
⚖️ Consumer Priorities - Control message distribution with priority consumers
🔧 Advanced Queue Options - Full control over queue arguments and behavior

Key Features
Requirements
Quick Start
Installation
Configuration
Usage
- Basic Usage
- Pool Monitoring
Advanced Features
Performance Tuning
Production Deployment
Testing
Troubleshooting
Contributing
License

Requirements

PHP 8.2 or higher
Laravel 11.x or 12.x
RabbitMQ Server 3.8+
ext-amqp PHP extension
ext-pcntl PHP extension (for parallel processing)

Quick Start

Get up and running in 5 minutes:

# 1. Install package
composer require iamfarhad/laravel-rabbitmq

# 2. Install AMQP extension (if not already installed)
pecl install amqp

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

# 4. Configure .env
echo "QUEUE_CONNECTION=rabbitmq" >> .env
echo "RABBITMQ_HOST=localhost" >> .env
echo "RABBITMQ_PORT=5672" >> .env
echo "RABBITMQ_USER=guest" >> .env
echo "RABBITMQ_PASSWORD=guest" >> .env

# 5. Create a job
php artisan make:job ProcessPodcast

# 6. Dispatch the job
php artisan tinker
>>> dispatch(new App\Jobs\ProcessPodcast());

# 7. Start worker
php artisan rabbitmq:consume --queue=default

That's it! Your jobs are now processing through RabbitMQ. 🎉

Installation

Install via Composer

composer require iamfarhad/laravel-rabbitmq

Install AMQP Extension

The package requires the ext-amqp PHP extension. Install it based on your environment:

macOS (Homebrew)

brew install rabbitmq-c
pecl install amqp

Ubuntu/Debian

sudo apt-get install librabbitmq-dev libssh-dev
sudo pecl install amqp

Docker

Add to your Dockerfile:

# Install dependencies
RUN apt-get update && apt-get install -y \
    librabbitmq-dev \
    libssh-dev

# Install AMQP extension
RUN pecl install amqp && docker-php-ext-enable amqp

Verify Installation

php -m | grep amqp

You should see amqp in the output.

Publish Configuration

php artisan vendor:publish --provider="iamfarhad\LaravelRabbitMQ\LaravelRabbitQueueServiceProvider" --tag="config"

Lumen Installation

$app->register(iamfarhad\LaravelRabbitMQ\LaravelRabbitQueueServiceProvider::class);

Configuration

Basic Configuration

Add the RabbitMQ connection to config/queue.php:

'connections' => [
    'rabbitmq' => [
        'driver' => 'rabbitmq',
        'queue'  => env('RABBITMQ_QUEUE', 'default'),
        
        'hosts' => [
            'host'      => env('RABBITMQ_HOST', '127.0.0.1'),
            'port'      => env('RABBITMQ_PORT', 5672),
            'user'      => env('RABBITMQ_USER', 'guest'),
            'password'  => env('RABBITMQ_PASSWORD', 'guest'),
            'vhost'     => env('RABBITMQ_VHOST', '/'),
            'heartbeat' => env('RABBITMQ_HEARTBEAT_CONNECTION', 0),
        ],
        
        // 🚀 Connection and Channel Pool Configuration
        'pool' => [
            // Connection Pool Settings
            'max_connections' => env('RABBITMQ_MAX_CONNECTIONS', 10),
            'min_connections' => env('RABBITMQ_MIN_CONNECTIONS', 2),
            
            // Channel Pool Settings  
            'max_channels_per_connection' => env('RABBITMQ_MAX_CHANNELS_PER_CONNECTION', 100),
            
            // Retry Strategy
            'max_retries' => env('RABBITMQ_MAX_RETRIES', 3),
            'retry_delay' => env('RABBITMQ_RETRY_DELAY', 1000), // milliseconds
            
            // Health Check Settings
            'health_check_enabled' => env('RABBITMQ_HEALTH_CHECK_ENABLED', true),
            'health_check_interval' => env('RABBITMQ_HEALTH_CHECK_INTERVAL', 30), // seconds
        ],
        
        'options' => [
            'ssl_options' => [
                'cafile'      => env('RABBITMQ_SSL_CAFILE'),
                'local_cert'  => env('RABBITMQ_SSL_LOCALCERT'),
                'local_key'   => env('RABBITMQ_SSL_LOCALKEY'),
                'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
                'passphrase'  => env('RABBITMQ_SSL_PASSPHRASE'),
            ],
            'queue' => [
                'job' => \iamfarhad\LaravelRabbitMQ\Jobs\RabbitMQJob::class,
                'qos' => [
                    'prefetch_size'  => 0,
                    'prefetch_count' => 10,
                    'global'         => false
                ]
            ],
        ],
    ],
]

Connection Pooling

The package features advanced connection and channel pooling for optimal performance:

Connection Pool Benefits

Resource Efficiency: Reuse existing connections instead of creating new ones
Performance: Significantly faster job processing with reduced connection overhead
Reliability: Automatic health monitoring and dead connection cleanup
Scalability: Configurable limits to handle varying workloads

Pool Configuration Options

| Setting | Default | Description | |---------|---------|-------------| | max_connections | 10 | Maximum number of connections in the pool | | min_connections | 2 | Minimum connections to maintain | | max_channels_per_connection | 100 | Maximum channels per connection | | max_retries | 3 | Connection retry attempts | | retry_delay | 1000ms | Initial retry delay (exponential backoff) | | health_check_enabled | true | Enable automatic health monitoring | | health_check_interval | 30s | Health check frequency |

Environment Variables

Add to your .env file:

# Basic RabbitMQ Configuration
QUEUE_CONNECTION=rabbitmq

# RabbitMQ Connection
RABBITMQ_HOST=127.0.0.1
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/
RABBITMQ_QUEUE=default

# Connection Options
RABBITMQ_HEARTBEAT_CONNECTION=0

# 🚀 Connection Pool Configuration
RABBITMQ_MAX_CONNECTIONS=10
RABBITMQ_MIN_CONNECTIONS=2
RABBITMQ_MAX_CHANNELS_PER_CONNECTION=100

# 🛡️ Retry Strategy
RABBITMQ_MAX_RETRIES=3
RABBITMQ_RETRY_DELAY=1000

# 📊 Health Monitoring
RABBITMQ_HEALTH_CHECK_ENABLED=true
RABBITMQ_HEALTH_CHECK_INTERVAL=30

# SSL/TLS Configuration (Optional)
RABBITMQ_SECURE=false
#RABBITMQ_SSL_CAFILE=/path/to/ca.pem
#RABBITMQ_SSL_LOCALCERT=/path/to/cert.pem
#RABBITMQ_SSL_LOCALKEY=/path/to/key.pem
#RABBITMQ_SSL_VERIFY_PEER=true

Usage

Basic Usage

Job Dispatching

// Dispatch to default queue
dispatch(new ProcessPodcast($podcast));

// Dispatch to specific queue
dispatch(new ProcessPodcast($podcast))->onQueue('podcasts');

// Delayed dispatch
dispatch(new ProcessPodcast($podcast))->delay(now()->addMinutes(10));

Creating Jobs

<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

class ProcessPodcast implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(
        public Podcast $podcast
    ) {}

    public function handle(): void
    {
        // Process

LaravelRabbitMQ

Install / Use

README