Laravel Rabbitmq Laravel Package

Getting Started

Minimal Steps

1. Installation:

   composer require iamfarhad/laravel-rabbitmq
   pecl install amqp
   

   Ensure ext-amqp is enabled in php.ini.

2. Publish Config:

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

3. Configure .env:

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

4. Dispatch a Job:

   dispatch(new App\Jobs\ProcessPodcast($podcast))->onQueue('podcasts');
   

5. Run Worker:

   php artisan rabbitmq:consume --queue=podcasts
   

First Use Case

Use this package for high-performance queue processing in production. Start with a simple job dispatch and worker consumption, then explore advanced features like quorum queues, publisher confirms, or multi-host failover as needed.

Implementation Patterns

Core Workflows

1. Job Dispatching:

   • Use Laravel’s native dispatch() syntax with optional queues/delays.
   • Example:

     dispatch(new SendEmailJob($user))->onQueue('emails')->delay(now()->addMinutes(5));
     

2. Worker Management:

   • Poll Mode (default): Safe for most use cases, matches Laravel’s worker lifecycle.

     php artisan rabbitmq:consume --queue=emails --consume-mode=poll
     

   • Consume Mode: High-performance for hot queues (one queue per worker recommended).

     php artisan rabbitmq:consume --queue=emails --consume-mode=consume
     

3. Multi-Host Failover: Configure multiple hosts in config/queue.php:

   'hosts' => [
       ['host' => 'rabbitmq-1', 'port' => 5672, ...],
       ['host' => 'rabbitmq-2', 'port' => 5672, ...],
   ],
   

   The package automatically balances connections across hosts.

4. Queue Topology: Define exchanges, routing keys, and queue types (e.g., quorum, priority) in config:

   'exchanges' => [
       'jobs' => [
           'type' => 'topic',
           'routing_key' => 'jobs.%s', // %s = queue name
       ],
   ],
   'queues' => [
       'critical' => ['priority' => 10],
       'quorum-orders' => ['quorum' => true],
   ],
   

5. Publisher Confirms: Enable for critical workflows where message receipt must be confirmed:

   RABBITMQ_PUBLISHER_CONFIRMS_ENABLED=true
   RABBITMQ_PUBLISHER_CONFIRMS_TIMEOUT=5
   

   Use in code:

   dispatch(new CriticalJob())->withPublisherConfirms();
   

6. Delayed Jobs: Leverage Laravel’s delay() or RabbitMQ’s delayed-message plugin:

   RABBITMQ_DELAYED_PLUGIN_ENABLED=true
   RABBITMQ_DELAYED_EXCHANGE=delayed
   

7. Dead-Letter Routing: Configure failed-job rerouting:

   RABBITMQ_REROUTE_FAILED=true
   RABBITMQ_FAILED_EXCHANGE=failed.jobs
   RABBITMQ_FAILED_ROUTING_KEY=%s.failed
   

Integration Tips

• Horizon: Enable with:

  RABBITMQ_WORKER=horizon
  

  Requires Horizon installed (laravel/horizon).

• Octane: Optimize pool reuse:

  RABBITMQ_OCTANE_RESET_ON_REQUEST=false  # Default (recommended)
  

• Admin Commands: Manage infrastructure via CLI:

  php artisan rabbitmq:exchange-declare jobs --type=topic
  php artisan rabbitmq:queue-declare orders --durable=1
  php artisan rabbitmq:pool-stats --watch
  

Gotchas and Tips

Pitfalls

1. Missing ext-amqp:

   • Error: Class AMQPConnection not found.
   • Fix: Install via pecl install amqp and enable in php.ini.

2. Parallel Workers Require pcntl:

   • Error: pcntl extension is required for parallel workers.
   • Fix: Install ext-pcntl or use --num-processes=1.

3. Horizon Events Not Triggering:

   • Ensure RABBITMQ_WORKER=horizon and Horizon is installed.

4. Quorum + Priority Conflicts:

   • Quorum queues cannot use priority. Configure one or the other per queue.

5. Connection Pool Exhaustion:

   • Monitor pool stats (php artisan rabbitmq:pool-stats) and adjust:

     RABBITMQ_MAX_CONNECTIONS=10
     RABBITMQ_MAX_CHANNELS_PER_CONNECTION=100
     

6. Delayed Jobs Without Plugin:

   • Without RABBITMQ_DELAYED_PLUGIN_ENABLED, uses TTL + dead-letter routing (less precise).

7. Consume Mode Scaling:

   • In consume mode, one queue per worker process is recommended. Scale horizontally with more workers/containers.

Debugging Tips

• Enable Verbose Logging:

  RABBITMQ_LOG_LEVEL=debug
  

  Logs appear in storage/logs/laravel-rabbitmq.log.

• Check Pool Health:

  php artisan rabbitmq:pool-stats --watch --interval=5
  

  Look for stalled connections or channel leaks.

• Inspect Raw Messages: Extend RabbitMQJob to access raw AMQP data:

  public function getRawBody() { ... }
  public function headers() { ... }
  

• Test Failover: Simulate host failures by stopping a RabbitMQ node and verifying workers reconnect.

Configuration Quirks

1. Lazy Connections:

   • Connections are created on-demand (useful for CLI/Octane). Disable with:

     'lazy' => false,
     

2. Transport Protocol:

   • Default: tcp. Use ssl/tls for secure connections:

     'transport' => 'tls',
     'port' => 5671, // Default TLS port
     

3. After-Commit Behavior:

   • Enable to dispatch jobs only after DB transactions commit:

     RABBITMQ_AFTER_COMMIT=true
     

4. Custom Job Classes:

   • Override RabbitMQJob for raw message handling:

     'options' => [
         'queue' => [
             'job' => App\Jobs\CustomRabbitMQJob::class,
         ],
     ],
     

Extension Points

1. Event Listeners:

   • Extend RabbitMQJob to hook into message lifecycle (e.g., reserved(), failed()).

2. Middleware:

   • Add middleware to jobs for pre/post-processing:

     public function handle() {
         // Custom logic
     }
     

3. Queue Builders:

   • Dynamically configure queues/exchanges at runtime:

     $queue = app(\iamfarhad\LaravelRabbitMQ\Queue\RabbitMQQueue::class)
         ->setQuorum(true)
         ->setPriority(5);
     

4. Custom Exchanges:

   • Declare exchanges dynamically:

     app(\iamfarhad\LaravelRabbitMQ\Exchange\RabbitMQExchange::class)
         ->declare('dynamic-exchange', 'direct', false, false, false);