Based on the php-amqplib/php-amqplib component, this libray adapts the AMQP protocol for Laravel, providing convenient and configurable queue work modes. Official documentation is available for further details. It supports mechanisms like delayed messages and dead-letter queues.

Environment Requirements:

• PHP version: ^7.3|^8.0

• Required extension: socket

• Additional setup:

  1. For delayed tasks, install the RabbitMQ delayed message plugin. For RabbitMQ 3.9.0:

  wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.9.0/rabbitmq_delayed_message_exchange-3.9.0.ez
  cp rabbitmq_delayed_message_exchange-3.9.0.ez /opt/rabbitmq/plugins/
  rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  
  

Usage:

Step 1: Install the Component

composer require sai97/laravel-amqp

Step 2: Publish Services and Configuration

php artisan vendor:publish --provider="Sai97\LaravelAmqp\AmqpQueueProviders"

This generates amqp.php (AMQP connection configuration) and app/QueueJob/DefaultQueueJob.php (default queue job) in the app/config directory.

amqp.php example:

<?php

use App\QueueJob\DefaultQueueJob;

return [
    "connection" => [
        "default" => [
            "host" => env("AMQP_HOST", "127.0.0.1"),
            "port" => env("AMQP_PORT", 5672),
            "user" => env("AMQP_USER", "root"),
            "password" => env("AMQP_PASSWORD", "root")
        ]
    ],

    "event" => [
        "default" => DefaultQueueJob::class,
    ]
];

The connection section configures AMQP connections, aligning with php-amqplib settings. The event section binds queue instances, using matching keys for easier management.

Supported Interface Methods:

// Get connection name
public function getConnectName(): string;

// Get exchange name
public function getExchangeName(): string;

// Get exchange type
public function getExchangeType(): string;

// Get queue name
public function getQueueName(): string;

// Get routing key
public function getRoutingKey(): string;

// Get content type
public function getContentType(): string;

// Enable dead-letter mode
public function isDeadLetter(): bool;

// Get dead-letter exchange name
public function getDeadLetterExchangeName(): string;

// Get dead-letter routing key
public function getDeadLetterRoutingKey(): string;

// Get dead-letter queue name
public function getDeadLetterQueueName(): string;

// Enable delayed tasks
public function isDelay(): bool;

// Get delay TTL
public function getDelayTTL(): int;

// Get queue arguments
public function getQueueArgs(): array;

// Get callback function
public function getCallback(): callable;

// Auto-acknowledge
public function isAutoAck(): bool;

Custom queue instances can be created by extending Sai97\LaravelAmqp\Queue. Refer to Sai97\LaravelAmqp\QueueInterface for configuration details.

Code Examples:

Prdoucer:

$payload = "This is a test message...";
$amqpService = new AmqpService(new BaseQueueJob);
$amqpService->produce($payload);

Consumer:

Define a custom command RabbitMQConsumer using Laravel's Command class. Configure queue bindings in amqp.php afterward. Example:

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Sai97\LaravelAmqp\AmqpService;
use Sai97\LaravelAmqp\QueueFactory;

class RabbitMQConsumer extends Command
{
    protected $signature = 'rabbitmq:consume {queue}';

    protected $description = 'RabbitMQ consumer process';

    public function __construct()
    {
        parent::__construct();
    }

    public function handle()
    {
        try {
            $queueName = $this->argument("queue");
            $queueConfig = config("amqp.event");

            if (!isset($queueConfig[$queueName]) || empty($entity = $queueConfig[$queueName])) {
                return $this->error("Unknown queue: {$queueName}");
            }

            $this->info("Starting RabbitMQ consumer for queue [{$queueName}]...");

            $amqpService = new AmqpService(new $entity);
            $amqpService->consume();

        } catch (\Throwable $e) {
            $queueName = $queueName ?? "";
            $this->error($e->getFile() . " [{$e->getLine()}]");
            return $this->error("Error in RabbitMQ consumer for queue [{$queueName}]: {$e->getMessage()}");
        }

        $this->info("Stopping RabbitMQ consumer for queue [{$queueName}]...");
    }
}

Run php artisan rabbitmq:consume default to start listening, where default matches the queue identifier in amqp.php.

Since consumers should run as daemons, use supervisord to manage processes. Example supervisord.conf:

[program:rabbitmq-consumer-default]
process_name=worker_%(process_num)d
numprocs=3
command=/usr/local/bin/php /app/www/laravel8/artisan rabbitmq:consume default
autostart=true
autorestart=true
startretries=3
priority=3
stdout_logfile=/var/log/rabbitmq-consumer-default.log
redirect_stderr=true

Supervisord simplifies management:

1. Add new queues by copying the program block and updating the configuration.
2. Use numprocs to scale consumers for parallel processing.

For supervisord operations, refer to its official documentation.